Portable Senders

[jackburridge]

At the moment each server will send, and receive on the same protocol, there should be a mechanism for portability. Each server should be able to receive the sender from another protocol type.

All senders should match the following type:

from collections.abc import Awaitable
from typing import AsyncContextManager
from collections.abc import Callable

from amgi_types import MessageSendEvent

_MessageSendManagerT = AsyncContextManager[
    Callable[[MessageSendEvent], Awaitable[None]]
]

Using an asynchronous context manager means we can leave the clean up to the manager itself, the server simply uses the returned sender

Todo

The following servers show have a message send implementation that can be shared

• amgi-redis
• amgi-aiokafka
• amgi-aiobotocore-sqs
• amgi-paho-mqtt
• amgi-sqs-event-source-mapping