libervia-backend/libervia/backend/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 libervia.backend.core import exceptions
from libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.core.i18n import _

class BaseApplicationHandler(abc.ABC):

    @abc.abstractmethod
    async def jingle_preflight(
        self,
        client: SatXMPPEntity,
        session: dict,
        description_elt: domish.Element
    ) -> None:
        """Called when preparation steps are needed by a plugin

        Notably used by XEP-0353 when a initiation message is received
        """
        pass

    @abc.abstractmethod
    async def jingle_preflight_info(
        self,
        client: SatXMPPEntity,
        session: dict,
        info_type: str,
        info_data: dict|None = None
    ) -> None:
        """Called when we have new information during preflight

        Notably used by XEP-0353 to signal ringing
        """
        pass

    @abc.abstractmethod
    async def jingle_preflight_cancel(
        self,
        client: SatXMPPEntity,
        session: dict,
        cancel_error: exceptions.CancelError
    ) -> None:
        """Called when preflight initiation is cancelled

        Notably used by XEP-0353 when an initiation message is cancelled
        """
        pass

    @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,
        client: SatXMPPEntity,
        action: str,
        session: dict,
        content_name: str,
        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):

    def is_usable(self, client, content_data: "ContentData") -> bool:
        """Return True if this transport is usable with given content

        This method provides a default implementation that always returns True. It can
        be overridden in subclasses to provide custom usability logic.

        @param client: SatXMPPEntity instance
        @param content_data: Jingle content data.
        @return: True if the transport is usable with this content, False otherwise
        """
        return True

    @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,
        reason_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 reason_elt: <reason> element
        """
        pass


@dataclass(frozen=True)
class ApplicationData:
    namespace: str
    handler: BaseApplicationHandler
    priority: int


@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