Mercurial > libervia-backend
view sat/plugins/plugin_xep_0166/models.py @ 4044:3900626bc100
plugin XEP-0166: refactoring, and various improvments:
- add models for transport and applications handlers and linked data
- split models into separate file
- some type hints
- some documentation comments
- add actions to prepare confirmation, useful to do initial parsing of all contents
- application arg/kwargs and some transport data can be initialised during Jingle
`initiate` call, this is notably useful when a call is made with transport data (this is
the call for A/V calls where codecs and ICE candidate can be specified when starting a
call)
- session data can be specified during Jingle `initiate` call
- new `store_in_session` argument in `_parse_elements`, which can be used to avoid
race-condition when a context element (<decription> or <transport>) is being parsed for
an action while an other action happens (like `transport-info`)
- don't sed `sid` in `transport_elt` during a `transport-info` action anymore in
`build_action`: this is specific to Jingle File Transfer and has been moved there
rel 419
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 15 May 2023 16:23:11 +0200 |
parents | |
children | 2ced30f6d5de |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia plugin for Jingle (XEP-0166) # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import abc from dataclasses import dataclass from typing import Awaitable, Callable, Union from twisted.internet import defer from twisted.words.xish import domish from sat.core.core_types import SatXMPPEntity from sat.core.i18n import _ class BaseApplicationHandler(abc.ABC): @abc.abstractmethod def jingle_request_confirmation( self, client: SatXMPPEntity, action: str, session: dict, content_name: str, desc_elt: domish.Element, ) -> Union[ Callable[..., Union[bool, defer.Deferred]], Callable[..., Awaitable[bool]] ]: """ If present, it is called on when session must be accepted. If not present, a generic accept dialog will be used. @param session: Jingle Session @param desc_elt: <description> element @return: True if the session is accepted. A Deferred can be returned. """ pass @abc.abstractmethod def jingle_session_init( self, client: SatXMPPEntity, session: dict, content_name: str, *args, **kwargs ) -> Union[ Callable[..., domish.Element], Callable[..., Awaitable[domish.Element]] ]: """ Must return the domish.Element used for initial content. @param client: SatXMPPEntity instance @param session: Jingle Session @param content_name: Name of the content @return: The domish.Element used for initial content """ pass @abc.abstractmethod def jingle_handler( self, client: SatXMPPEntity, action: str, session: dict, content_name: str, transport_elt: domish.Element ) -> Union[ Callable[..., None], Callable[..., Awaitable[None]] ]: """ Called on several actions to negotiate the application or transport. @param client: SatXMPPEntity instance @param action: Jingle action @param session: Jingle Session @param content_name: Name of the content @param transport_elt: Transport element """ pass @abc.abstractmethod def jingle_terminate( self, reason_elt: domish.Element ) -> Union[ Callable[..., None], Callable[..., Awaitable[None]] ]: """ Called on session terminate, with reason_elt. May be used to clean session. @param reason_elt: Reason element """ pass class BaseTransportHandler(abc.ABC): @abc.abstractmethod def jingle_session_init( self, client: SatXMPPEntity, session: dict, content_name: str, *args, **kwargs ) -> Union[ Callable[..., domish.Element], Callable[..., Awaitable[domish.Element]] ]: """ Must return the domish.Element used for initial content. @param client: SatXMPPEntity instance @param session: Jingle Session @param content_name: Name of the content @return: The domish.Element used for initial content """ pass @abc.abstractmethod def jingle_handler( self, client: SatXMPPEntity, action: str, session: dict, content_name: str, transport_elt: domish.Element ) -> Union[ Callable[..., None], Callable[..., Awaitable[None]] ]: """ Called on several actions to negotiate the application or transport. @param client: SatXMPPEntity instance @param action: Jingle action @param session: Jingle Session @param content_name: Name of the content @param transport_elt: Transport element """ pass @dataclass(frozen=True) class ApplicationData: namespace: str handler: BaseApplicationHandler @dataclass(frozen=True) class TransportData: namespace: str handler: BaseTransportHandler priority: int @dataclass(frozen=True) class ContentData: application: ApplicationData app_args: list app_kwargs: dict transport_data: dict content_name: str