Source code for aries_cloudagent.transport.inbound.base

"""Base inbound transport class."""

from abc import ABC, abstractmethod
from collections import namedtuple

from ...error import BaseError


[docs]class BaseInboundTransport(ABC):
    """Base inbound transport class."""

[docs]    @abstractmethod
    async def start(self) -> None:
        """Start listening for on this transport."""

[docs]    @abstractmethod
    async def stop(self) -> None:
        """Stop listening for on this transport."""


[docs]class InboundTransportRegistrationError(BaseError):
    """Error in loading an inbound transport."""


[docs]class InboundTransportSetupError(BaseError):
    """Setup error for an inbound transport."""


InboundTransportConfiguration = namedtuple(
    "InboundTransportConfiguration", "module host port"
)