view libervia/backend/plugins/plugin_xep_0166/models.py @ 4100:810921c33a47

tools (common/template): add filter to get media types: Add 2 filters to get main type and subtype of media type. Jinja2 and Nunjucks don't handle slices in the same way (Python way for Jinja2, JS way for Nunjucks), making it difficult to retrieve main type of a media from media type. Thoses filters work in both cases.
author Goffi <goffi@goffi.org>
date Thu, 22 Jun 2023 15:49:06 +0200
parents 4b842c1fb686
children bc60875cb3b8
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 libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.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,
        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):

    @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


@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