erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0167: handle conferences:

Browse source 
- SDP can now be answered by component instead of frontend. This is useful for A/V
  conferences component to handle A/V call jingle sessions.
- new `call_update` and method, and `content-add` action preparation. This is not yet used
  by A/V conference, but it's a preparation for a potential future use.
- Add NS_AV_CONFERENCES to features as required by the newly proposed A/V Conferences
  protoXEP.

rel 447
This commit is contained in:
Goffi 2024-07-29 03:31:09 +02:00
commit 96a1f4435c
2 changed files with 130 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

140
libervia/backend/plugins/plugin_xep_0167/__init__.py
View file

@ -94,13 +94,21 @@ class XEP_0167(BaseApplicationHandler):
method=self._call_start,
async_=True,
)
host.bridge.add_method(
"call_update",
".plugin",
in_sign="sss",
out_sign="",
method=self._call_update,
async_=True,
)
host.bridge.add_method(
"call_answer_sdp",
".plugin",
in_sign="sss",
out_sign="",
method=self._call_answer_sdp,
async_=True,
async_=False,
)
host.bridge.add_method(
"call_info",
@ -122,6 +130,9 @@ class XEP_0167(BaseApplicationHandler):
# profile
host.bridge.add_signal("call_setup", ".plugin", signature="sss")
# args: session_id, serialised update data, profile
host.bridge.add_signal("call_update", ".plugin", signature="sss")
# args: session_id, data, profile
host.bridge.add_signal("call_ended", ".plugin", signature="sss")
@ -146,6 +157,17 @@ class XEP_0167(BaseApplicationHandler):
)
)
def _call_update(
self,
session_id: str,
call_data_s: str,
profile_key: str,
):
client = self.host.get_client(profile_key)
return defer.ensureDeferred(
self.call_update(client, session_id, data_format.deserialise(call_data_s))
)
def parse_call_data(self, call_data: dict) -> dict:
"""Parse ``call_data`` and return corresponding contents end metadata"""
metadata = call_data.get("metadata") or {}
@ -233,6 +255,7 @@ class XEP_0167(BaseApplicationHandler):
client: SatXMPPEntity,
peer_jid: jid.JID,
call_data: dict,
session_id: str | None = None,
) -> str:
"""Initiate a call session with the given peer.
@ -247,6 +270,8 @@ class XEP_0167(BaseApplicationHandler):
- id (str): Identifier for the media (optional).
- ice-candidates: ICE candidates for media transport.
- And other transport specific data.
@param session_id: ID of the Jingle session. If None, an ID will be automatically
generated.
@return: Session ID (SID) for the initiated call session.
@ -271,9 +296,75 @@ class XEP_0167(BaseApplicationHandler):
call_type=call_type,
metadata=metadata,
peer_metadata={},
sid=session_id,
)
return sid
async def call_update(
self,
client: SatXMPPEntity,
session_id: str,
call_data: dict,
) -> None:
"""Update a running call session.
@param session_id: ID of the Jingle session to update.
@param call_data: Dictionary containing updated data for the call. Must include SDP information.
The dict can have the following keys:
- sdp (str): SDP data for the call.
- metadata (dict): Additional metadata for the call (optional).
Each media type ("audio" and "video") in the SDP should have:
- application_data (dict): Data about the media.
- fingerprint (str): Security fingerprint data (optional).
- id (str): Identifier for the media (optional).
- ice-candidates: ICE candidates for media transport.
- And other transport specific data.
@raises exceptions.DataError: If media data is invalid or duplicate content name
(mid) is found.
"""
session = self._j.get_session(client, session_id)
try:
new_offer_sdp = call_data["sdp"]
except KeyError:
raise exceptions.DataError(f"New SDP offer is missing: {call_data}")
metadata = self.parse_call_data(call_data)
contents = self.get_contents(call_data, metadata)
if not contents:
raise exceptions.DataError("no valid media data found: {call_data}")
call_type = (
C.META_SUBTYPE_CALL_VIDEO
if "video" in call_data
else C.META_SUBTYPE_CALL_AUDIO
)
for content_args in contents:
content = content_args["app_kwargs"]
content["app_ns"] = NS_JINGLE_RTP
content["name"] = (content_args["name"],)
content["transport_type"] = self._j.TRANSPORT_DATAGRAM
media = content["media"]
media_data = content["media_data"].copy()
media_data["transport_data"] = content_args["transport_data"][
"local_ice_data"
]
desc_elt = mapping.build_description(media, media_data, {})
iq_elt, __ = self._j.build_action(
client,
self._j.A_CONTENT_ADD,
session,
content_args["name"],
context_elt=desc_elt,
)
content_data = self._j.get_content_data(content)
transport = self._j.get_transport(client, content, content_data)
transport_elt = transport.handler.build_transport(
media_data["transport_data"]
)
iq_elt.jingle.content.addChild(transport_elt)
await iq_elt.send()
def _call_answer_sdp(self, session_id: str, answer_sdp: str, profile: str) -> None:
client = self.host.get_client(profile)
session = self._j.get_session(client, session_id)
@ -551,26 +642,43 @@ class XEP_0167(BaseApplicationHandler):
local_ice_data = media_data["transport_data"]
transport_data["local_ice_data"] = local_ice_data
def send_answer_sdp(self, client: SatXMPPEntity, session: dict) -> None:
async def send_answer_sdp(self, client: SatXMPPEntity, session: dict) -> None:
"""Send answer SDP to frontend"""
if not session.get(ANSWER_SDP_SENT_KEY, False):
# we only send the signal once, as it means that the whole session is
# accepted
answer_sdp = mapping.generate_sdp_from_session(session)
self.host.bridge.call_setup(
session["id"],
data_format.serialise(
call_setup = session.get("call_setup_cb")
if call_setup is None:
self.host.bridge.call_setup(
session["id"],
data_format.serialise(
{
"role": session["role"],
"sdp": answer_sdp,
}
),
client.profile,
)
else:
await call_setup(
client,
session,
{
"role": session["role"],
"sdp": answer_sdp,
}
),
client.profile,
)
},
)
session[ANSWER_SDP_SENT_KEY] = True
async def jingle_handler(self, client, action, session, content_name, desc_elt):
content_data = session["contents"][content_name]
if action == self._j.A_CONTENT_ADD:
content_data = session["contents_new"][content_name]
else:
content_data = session["contents"][content_name]
application_data = content_data["application_data"]
if action == self._j.A_PREPARE_CONFIRMATION:
session["metadata"] = {}
@ -592,7 +700,16 @@ class XEP_0167(BaseApplicationHandler):
elif action == self._j.A_PREPARE_INITIATOR:
application_data["peer_data"] = mapping.parse_description(desc_elt)
elif action == self._j.A_SESSION_ACCEPT:
self.send_answer_sdp(client, session)
await self.send_answer_sdp(client, session)
elif action == self._j.A_CONTENT_ADD:
current_contents = session["contents"]
if content_name in current_contents:
raise exceptions.ConflictError(
f"There is already a {content_name!r} content."
)
current_contents[content_name] = content_data
application_data["media"] = desc_elt["media"]
application_data["peer_data"] = mapping.parse_description(desc_elt)
else:
log.warning(f"FIXME: unmanaged action {action}")
@ -678,6 +795,7 @@ class XEP_0167_handler(XMPPHandler):
disco.DiscoFeature(NS_JINGLE_RTP),
disco.DiscoFeature(NS_JINGLE_RTP_AUDIO),
disco.DiscoFeature(NS_JINGLE_RTP_VIDEO),
disco.DiscoFeature(NS_AV_CONFERENCES),
]
def getDiscoItems(self, requestor, target, nodeIdentifier=""):

2
libervia/backend/plugins/plugin_xep_0343.py
View file

@ -313,7 +313,7 @@ class XEP_0343(BaseTransportHandler):
# we should have the answer long before 2 min
answer_sdp_d.addTimeout(2 * 60, reactor)
self._rtp.send_answer_sdp(client, session)
await self._rtp.send_answer_sdp(client, session)
answer_sdp = await answer_sdp_d
parsed_answer = mapping.parse_sdp(answer_sdp, session["role"])