py-libp2p/libp2p/protocol_muxer/multiselect_muxer_interface.py

from abc import (
    ABC,
    abstractmethod,
)
from typing import (
    Dict,
    Tuple,
)

from libp2p.typing import (
    StreamHandlerFn,
    TProtocol,
)

from .multiselect_communicator_interface import (
    IMultiselectCommunicator,
)


class IMultiselectMuxer(ABC):
    """
    Multiselect module that is responsible for responding to a multiselect
    client and deciding on a specific protocol and handler pair to use for
    communication.
    """

    handlers: Dict[TProtocol, StreamHandlerFn]

    @abstractmethod
    def add_handler(self, protocol: TProtocol, handler: StreamHandlerFn) -> None:
        """
        Store the handler with the given protocol.

        :param protocol: protocol name
        :param handler: handler function
        """

    def get_protocols(self) -> Tuple[TProtocol, ...]:
        return tuple(self.handlers.keys())

    @abstractmethod
    async def negotiate(
        self, communicator: IMultiselectCommunicator
    ) -> Tuple[TProtocol, StreamHandlerFn]:
        """
        Negotiate performs protocol selection.

        :param stream: stream to negotiate on
        :return: selected protocol name, handler function
        :raise Exception: negotiation failed exception
        """