Message bus

Expanse comes with a built-in message bus that allows different components of the system to communicate with each other in a decoupled way.

The general idea is that the application can dispatch messages, and message handlers can process them asynchronously.

The message bus can be used for a variety of use cases and is what powers Expanse's own queue system.

Creating messages and handlers

A message is a simple class that can be serialized to JSON. The message bus natively supports dataclasses, Pydantic models and msgspec structs.

from dataclasses import dataclass


@dataclass
class UserCreated:

    user_id: int
    username: str

from pydantic import BaseModel


class UserCreated(BaseModel):

    user_id: int
    username: str

import msgspec


class UserCreated(msgspec.Struct):

    user_id: int
    username: str

A message handler is any callable that takes a message as an argument, marked with the @message_handler decorator.

from expanse.messenger.utils import message_handler

from app.messages.user_created import UserCreated


@message_handler()
async def send_welcome_email(message: UserCreated) -> None:
    ...

By default, Expanse will look for message handlers in the app/message_handlers directory.

Dispatching messages

To dispatch a message, you may inject an instance of the MessageBus and call the dispatch method.

from expanse.messenger.asynchronous.message_bus import MessageBus

from app.messages.user_created import UserCreated


class UserController:

    async def create_user(self, bus: MessageBus) -> Response:
        ...
        await bus.dispatch(UserCreated(user_id=1, username="john_doe"))

        return Response(status_code=201)
        ...

from expanse.messenger.synchronous.message_bus import MessageBus

from app.messages.user_created import UserCreated


class UserController:

    def create_user(self, bus: MessageBus) -> Response:
        ...
        bus.dispatch(UserCreated(user_id=1, username="john_doe"))

        return Response(status_code=201)
        ...

Dispatching messages asynchronously using transports

By default, the message bus dispatches messages synchronously, meaning that the message handlers are executed in the same process as the code that dispatches the message. If you want to dispatch messages asynchronously, you can configure a transport for the message bus. Expanse supports multiple transports out of the box.

Transports can be registered via dedicated environment variables:

EXPANSE_MESSENGER_TRANSPORT=async
EXPANSE_MESSENGER_TRANSPORT__ASYNC__DRIVER=database
EXPANSE_MESSENGER_TRANSPORT__ASYNC__CONNECTION=sqlite

Here we are configuring a transport named async that uses the database driver and using the sqlite database connection.

Dispatch and database transactions

If you are dispatching messages while a database transaction is active, via a Session instance, the messages will not be dispatched until the transaction is committed. This ensures that if the transaction is rolled back, the messages will not be dispatched either, preventing potential inconsistencies between the state of the database and the messages being processed.

This also works for nested transactions: only messages dispatched during committed nested transactions will be dispatched when the main transaction is committed.

Consuming messages

Dispatched messages need to by consumed to be processed by their handlers. To consume messages, you can run the messenger consume command:

python ./beam messenger consume async

This command will start a worker that listens for new messages on the async transport and processes them using their handlers.

If you want the worker to automatically stop after processing a certain number of messages, you can use the --limit option:

python ./beam messenger consume async --limit 100

Managing failures

If an error occurs while processing a message, it will be automatically re-sent to the transport to be tried again. By default, a message will be retried 3 times before being discarded or sent to the configured failure transport. To account for transient errors, each retry is delayed.

Retry strategies

How retries are handled can be configured using retry strategies. Expanse comes with a built-in multiplier strategy that increases the delay between retries exponentially.

MESSENGER_RETRY_STRATEGIES__MY_STRATEGY__TYPE=multiplier
# Maximum number of times a message will be retried
MESSENGER_RETRY_STRATEGIES__MY_STRATEGY__MAX_RETRIES=3
# Time to wait before the first retry (in milliseconds)
MESSENGER_RETRY_STRATEGIES__MY_STRATEGY__DELAY=1000
# Maximum delay between retries (in milliseconds)
MESSENGER_RETRY_STRATEGIES__MY_STRATEGY__MAX_DELAY=
# Multiplier applied to the delay on each subsequent retry
MESSENGER_RETRY_STRATEGIES__MY_STRATEGY__MULTIPLIER=2
# Randomness factor (between 0 and 1) added to each delay to
# prevent a thundering herd effect upon multiple messages being retried
MESSENGER_RETRY_STRATEGIES__MY_STRATEGY__JITTER=0.1

If you define a custom retry strategy, you can use it in your transport configuration:

EXPANSE_MESSENGER_TRANSPORT__ASYNC__RETRY_STRATEGY=my_strategy

Preventing retries

If you want to prevent a message from being retried after a failure, you can raise the UnrecoverableMessageHandlingError exception in the message handler:

from expanse.messenger.exceptions import UnrecoverableMessageHandlingError


@message_handler()
async def send_welcome_email(message: UserCreated) -> None:
    ...
    if error_is_unrecoverable:
        raise UnrecoverableMessageHandlingError("This message cannot be retried")

Saving failed messages

If you want to keep track of failed messages, you can configure a failure transport. Expanse supports the same drivers for failure transports as for regular transports.

EXPANSE_MESSENGER_FAILURE_TRANSPORT=failed
EXPANSE_MESSENGER_TRANSPORT__FAILED__DRIVER=database
EXPANSE_MESSENGER_TRANSPORT__FAILED__CONNECTION=sqlite

Making handlers idempotent

Since there is a possibility that the same message is delivered more than once, even under normal circumstances (e.g. if the worker is restarted while processing a message), it's a good practice to make message handlers idempotent. This means that processing the same message multiple times should have the same effect as processing it once.

Configuring transports

Database transport

The database transport uses a database table to store messages.

Before using the database transport, you need to set up a database connection in your database configuration.

Options:

• driver (mandatory): The driver to use for the transport. It should be set to database.

  MESSENGER_TRANSPORT__ASYNC__DRIVER=database
  

• connection (mandatory): The name of the database connection to use (as defined in the database configuration).

  MESSENGER_TRANSPORT__ASYNC__CONNECTION=sqlite
  

• table_name (optional): The name of the table where messages will be stored. Default is messages.

  MESSENGER_TRANSPORT__ASYNC__TABLE_NAME=messages
  

• queue_name (optional): The name of the queue to use. Default is default.

  MESSENGER_TRANSPORT__ASYNC__QUEUE_NAME=default
  

• redelivery_timeout (optional): Timeout (in seconds) before redelivering messages still in handling state (i.e: delivered_at is not null and message is still in table). Default is 3600 seconds.

  MESSENGER_TRANSPORT__ASYNC__REDELIVERY_TIMEOUT=3600
  

Setup

You will need to create the database table for the transport using the make messages table command:

python ./beam messenger make messages table

If you want to use another name for the table, you can use --table-name option when running the command:

python ./beam messenger make messages table --table-name my_messages

Redis transport

The Redis transport uses streams to store messages.

Before using the Redis transport, you need to set up a Redis connection in your Redis configuration.

Options

• driver (mandatory): The driver to use for the transport. It should be set to redis.

  MESSENGER_TRANSPORT__ASYNC__DRIVER=redis
  

• connection (mandatory): The name of the Redis connection to use (as defined in the Redis configuration).

  MESSENGER_TRANSPORT__ASYNC__CONNECTION=redis
  

• stream (mandatory): The name of the stream where messages will be stored.

  MESSENGER_TRANSPORT__ASYNC__STREAM=messages
  

• group (mandatory): The name of the consumer group to use.

  MESSENGER_TRANSPORT__ASYNC__GROUP=expanse
  

• consumer (mandatory): The name of the consumer to use for retrieving messages from the stream. This is used in combination with the group name to identify the consumer in the consumer group. It should be unique for each instance of the messenger to avoid conflicts with other instances consuming from the same stream and group.

  MESSENGER_TRANSPORT__ASYNC__CONSUMER=worker-1
  

Memory transport

The memory transport does not actually dispatch messages anywhere, but instead keeps them in memory. This can be useful for testing purposes.

Envelopes and stamps

When a message is dispatched, it is wrapped in an envelope that contains the message itself and additional metadata called stamps. Stamps can be used to store information about the message, such as the time it was dispatched, the number of times it has been retried, etc. They are heavily used internally by the message bus but you can add your own stamps if needed. Similarly to messages, stamps can be implemented as dataclasses, Pydantic models or msgspec structs.

from dataclasses import dataclass


@dataclass
class MyCustomStamp:
    data: str

from pydantic import BaseModel


class MyCustomStamp(BaseModel):
    data: str

import msgspec

class MyCustomStamp(msgspec.Struct):
    data: str

You can specify custom stamps to be added to the envelope when dispatching a message by passing them as a list to the dispatch method:

from expanse.messenger.asynchronous.message_bus import MessageBus
from app.messages.user_created import UserCreated
from app.messages.stamps.my_custom_stamp import MyCustomStamp


class UserController:

    async def create_user(self, bus: MessageBus) -> Response:
        ...
        await bus.dispatch(
            UserCreated(user_id=1, username="john_doe"),
            stamps=[MyCustomStamp(data="some data")]
        )

        return Response(status_code=201)
        ...

from expanse.messenger.synchronous.message_bus import MessageBus
from app.messages.user_created import UserCreated
from app.messages.stamps.my_custom_stamp import MyCustomStamp


class UserController:

    def create_user(self, bus: MessageBus) -> Response:
        ...
        bus.dispatch(
            UserCreated(user_id=1, username="john_doe"),
            stamps=[MyCustomStamp(data="some data")]
        )

        return Response(status_code=201)
        ...

Alternatively, you can also dispatch an already stamped envelope by passing it directly to the dispatch method:

message = UserCreated(user_id=1, username="john_doe")
envelope = Envelope.wrap(message).with_stamps(MyCustomStamp(data="some data"))
await bus.dispatch(envelope)

message = UserCreated(user_id=1, username="john_doe")
envelope = Envelope.wrap(message).with_stamps(MyCustomStamp(data="some data"))
bus.dispatch(envelope)

Middleware

When a message is dispatched or received, it goes through a stack of middleware. Each middleware can inspect or modify the envelope and decide whether to pass it to the next middleware in the stack or not.

A middleware is a class that defines a handle() method which accepts an Envelope instance and a callable responsible for calling the next middleware in the stack.

from collections.abc import Awaitable, Callable

from expanse.messenger.envelope import Envelope


class MyCustomMiddleware:

    async def handle(
        self,
        envelope: Envelope,
        next_call: Callable[[Envelope], Awaitable[Envelope]],
    ) -> Envelope:
        # Do something with the envelope
        # before it is passed to the next middleware or message handler
        ...

        await next_call(envelope)

        # Do something with the envelope
        # after it has been processed by the message handler
        # or processed by the next middleware in the stack
        ...

Note that the same middleware class can be used for both dispatching and receiving messages, so if you need to differentiate the behavior based on whether the message is being dispatched or received, you can check whether the envelope has a ReceivedStamp or not:

from collections.abc import Awaitable, Callable

from expanse.messenger.envelope import Envelope
from expanse.messenger.stamps.received import ReceivedStamp


class MyCustomMiddleware:

    async def handle(
        self,
        envelope: Envelope,
        next_call: Callable[[Envelope], Awaitable[Envelope]]
    ) -> Envelope:
        if envelope.has_stamp(ReceivedStamp):
            # The message is being received
            ...
        else:
            # The message is being dispatched
            ...

        await next_call(envelope)

To register a middleware, you can add it to the stack. This is typically done in a service provider:

from expanse.messenger.middleware.middleware_stack import MiddlewareStack


class AppServiceProvider(ServiceProvider):

    async def boot(self) -> None:
        self._container.on_resolved(MiddlewareStack, self._register_middleware)

    def _register_middleware(self, stack: MiddlewareStack) -> None:
        stack.append(MyCustomMiddleware)