libervia-backend/sat/plugins/plugin_xep_0166/models.py

#!/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