Mercurial > libervia-backend
changeset 4276:00a9316547ed
plugin XEP-0167: move ICE candidates parsing/SDP line building in separate methods:
These methods will be used by incoming "Conferences" component.
rel 445
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 05 Jul 2024 17:18:37 +0200 |
parents | 29fda1340078 |
children | b4b4ea8c5c87 |
files | libervia/backend/plugins/plugin_xep_0167/mapping.py |
diffstat | 1 files changed, 63 insertions(+), 44 deletions(-) [+] |
line wrap: on
line diff
--- a/libervia/backend/plugins/plugin_xep_0167/mapping.py Fri Jul 05 17:18:32 2024 +0200 +++ b/libervia/backend/plugins/plugin_xep_0167/mapping.py Fri Jul 05 17:18:37 2024 +0200 @@ -183,31 +183,8 @@ sdp_lines.append(f"a=ice-pwd:{ice_data['pwd']}") for candidate in ice_data["candidates"]: - foundation = candidate["foundation"] - component_id = candidate["component_id"] - transport = candidate["transport"] - priority = candidate["priority"] - address = candidate["address"] - candidate_port = candidate["port"] - candidate_type = candidate["type"] - - candidate_line = ( - f"a=candidate:{foundation} {component_id} {transport} {priority} " - f"{address} {candidate_port} typ {candidate_type}" - ) - - if "rel_addr" in candidate and "rel_port" in candidate: - candidate_line += ( - f" raddr {candidate['rel_addr']} rport {candidate['rel_port']}" - ) - - if "generation" in candidate: - candidate_line += f" generation {candidate['generation']}" - - if "network" in candidate: - candidate_line += f" network {candidate['network']}" - - sdp_lines.append(candidate_line) + candidate_line = generate_candidate_line(candidate) + sdp_lines.append(f"a={candidate_line}") # Generate a= lines for encryption if "encryption" in media_data: @@ -240,6 +217,66 @@ return "\r\n".join(sdp_lines) + "\r\n" +def generate_candidate_line(candidate: dict) -> str: + """Generate a ``candidate:`` attribute line from candidate data. + + @param candidate: ICE candidate data. + @return ICE candidate attribute line. + """ + foundation = candidate["foundation"] + component_id = candidate["component_id"] + transport = candidate["transport"] + priority = candidate["priority"] + address = candidate["address"] + candidate_port = candidate["port"] + candidate_type = candidate["type"] + + candidate_line = ( + f"candidate:{foundation} {component_id} {transport} {priority} " + f"{address} {candidate_port} typ {candidate_type}" + ) + + if "rel_addr" in candidate and "rel_port" in candidate: + candidate_line += f" raddr {candidate['rel_addr']} rport {candidate['rel_port']}" + + if "generation" in candidate: + candidate_line += f" generation {candidate['generation']}" + + if "network" in candidate: + candidate_line += f" network {candidate['network']}" + + return candidate_line + + +def parse_candidate(parts: list[str]) -> dict: + """Parse parts of a ICE candidate + + @param parts: Parts of the candidate line. + @return: candidate data + """ + candidate = { + "foundation": parts[0], + "component_id": int(parts[1]), + "transport": parts[2], + "priority": int(parts[3]), + "address": parts[4], + "port": int(parts[5]), + "type": parts[7], + } + + for part in parts[8:]: + if part == "raddr": + candidate["rel_addr"] = parts[parts.index(part) + 1] + elif part == "rport": + candidate["rel_port"] = int(parts[parts.index(part) + 1]) + elif part == "generation": + candidate["generation"] = parts[parts.index(part) + 1] + elif part == "network": + candidate["network"] = parts[parts.index(part) + 1] + + return candidate + + def parse_sdp(sdp: str, role: str) -> dict: """Parse SDP string. @@ -377,25 +414,7 @@ elif attribute == "candidate": assert transport_data is not None - candidate = { - "foundation": parts[0], - "component_id": int(parts[1]), - "transport": parts[2], - "priority": int(parts[3]), - "address": parts[4], - "port": int(parts[5]), - "type": parts[7], - } - - for part in parts[8:]: - if part == "raddr": - candidate["rel_addr"] = parts[parts.index(part) + 1] - elif part == "rport": - candidate["rel_port"] = int(parts[parts.index(part) + 1]) - elif part == "generation": - candidate["generation"] = parts[parts.index(part) + 1] - elif part == "network": - candidate["network"] = parts[parts.index(part) + 1] + candidate = parse_candidate(parts) transport_data.setdefault("candidates", []).append(candidate)