Skip to content

Transport

Bases: ABC

Abstract base for any transport. It should manage publishing, subscribing, connecting, and disconnecting.

Source code in eggai/transport/base.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class Transport(ABC):
    """
    Abstract base for any transport. It should manage publishing,
    subscribing, connecting, and disconnecting.
    """

    @abstractmethod
    async def connect(self):
        """
        Connect to the underlying system.
        """
        pass

    @abstractmethod
    async def disconnect(self):
        """
        Cleanly disconnect from the transport.
        """
        pass

    @abstractmethod
    async def publish(self, channel: str, message: Union[Dict[str, Any], BaseModel]):
        """
        Publish the given message to the channel.
        """
        pass

    @abstractmethod
    async def subscribe(self, channel: str, callback: Callable[[Dict[str, Any]], "asyncio.Future"], **kwargs) -> Callable:
        """
        Subscribe to a channel with the given callback, invoked on new messages.
        (No-op if a consumer doesn’t exist.)
        """
        pass

connect() abstractmethod async

Connect to the underlying system.

Source code in eggai/transport/base.py 
16
17
18
19
20
21
@abstractmethod
async def connect(self):
    """
    Connect to the underlying system.
    """
    pass

disconnect() abstractmethod async

Cleanly disconnect from the transport.

Source code in eggai/transport/base.py 
23
24
25
26
27
28
@abstractmethod
async def disconnect(self):
    """
    Cleanly disconnect from the transport.
    """
    pass

publish(channel, message) abstractmethod async

Publish the given message to the channel.

Source code in eggai/transport/base.py 
30
31
32
33
34
35
@abstractmethod
async def publish(self, channel: str, message: Union[Dict[str, Any], BaseModel]):
    """
    Publish the given message to the channel.
    """
    pass

subscribe(channel, callback, **kwargs) abstractmethod async

Subscribe to a channel with the given callback, invoked on new messages. (No-op if a consumer doesn’t exist.)

Source code in eggai/transport/base.py 
37
38
39
40
41
42
43
@abstractmethod
async def subscribe(self, channel: str, callback: Callable[[Dict[str, Any]], "asyncio.Future"], **kwargs) -> Callable:
    """
    Subscribe to a channel with the given callback, invoked on new messages.
    (No-op if a consumer doesn’t exist.)
    """
    pass