RabbitMQ Broker Events#

Use broker event callbacks to observe the RabbitMQ connection: log failures, record metrics, or signal a change in availability to other parts of your application.

Pass sequences of callbacks as keyword arguments when creating RabbitBroker. FastStream registers each on_close callback in aio-pika's close_callbacks collection and each on_reconnected callback in its reconnect_callbacks collection. See the official connection implementation and robust connection implementation for the underlying events.

Register callbacks#

Both arguments accept a sequence of callbacks, such as a list or tuple, and default to None. Pass even a single callback inside a sequence. Each callback can be a sync or async function returning None, and a sequence can contain both. Use multiple callbacks to keep actions such as logging and metrics separate. The example uses one async function for each event.

Callback execution order is not guaranteed because aio-pika stores callbacks in a set. Register only the events your application needs.

 1
 2
 3
 4
 5
 6
 7
 8
 9
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

import logging

from aio_pika.abc import AbstractConnection, AbstractRobustConnection

from faststream import FastStream, Logger
from faststream.rabbit import RabbitBroker

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rabbit.events")


async def on_close(
    connection: AbstractConnection | None,
    error: BaseException | None,
) -> None:
    if connection is not None and connection.close_called:
        logger.info("RabbitMQ connection closed during shutdown")
    else:
        logger.warning("RabbitMQ connection closed: %r", error)


async def on_reconnected(connection: AbstractRobustConnection | None) -> None:
    logger.info("RabbitMQ connection restored")


broker = RabbitBroker(
    "amqp://guest:guest@localhost:5672/",
    on_close=[on_close],
    on_reconnected=[on_reconnected],
)
app = FastStream(broker)


@broker.subscriber("events.example")
async def handle(message: str, logger: Logger) -> None:
    logger.info("Received: %s", message)

With a RabbitMQ server running locally, save this example as broker_events.py and run:

faststream run broker_events:app

To observe closure and reconnection, wait for the application to start, stop RabbitMQ, and restart it while the application is still running. Stop the application with Ctrl+C to observe closure during normal shutdown.

Callback reference#

The connection argument can be None according to aio-pika's typing contract. Its callback collections hold a weak reference to the sender. See the official callback types and collections.

Closure: on_close#

Use this callback to record connection loss, for example after a server restart or a network failure. It also runs when FastStream stops the broker deliberately. The second argument is the closing exception, which can be None. Its type is BaseException because closure can involve cancellation.

A close notification does not mean that automatic reconnection has stopped. aio-pika can recover from temporary connection loss. Check connection.close_called, when the connection argument is available, to distinguish an explicit close from an unexpected interruption. A closing exception alone does not identify an unexpected failure.

aio-pika can also invoke this callback if channel restoration fails. Treat it as a connection notification rather than an application exception handler. Handle exceptions from publishing or requests around the operation. Use ExceptionMiddleware for FastStream message-processing error handling.

Reconnection: on_reconnected#

Use this callback to record recovery after an interruption. During automatic recovery, aio-pika invokes it after restoring the connection and its robust channels. You do not need to call broker.connect() or register your subscribers again inside the callback.

This callback does not run on the first successful connection. For application initialization after the broker starts, use FastStream's after_startup hook.

Reconnection and application behavior#

RabbitBroker uses aio-pika's robust connection to reconnect automatically. reconnect_interval controls the delay between attempts and defaults to 5.0 seconds. Registering callbacks does not change the retry behavior. fail_fast defaults to True and controls whether the initial connection attempt fails immediately when RabbitMQ is unavailable.

FastStream registers the callbacks after establishing a connection, including each new connection created after a broker restart. These callbacks do not report failures during initial connection establishment. Handle those errors around broker startup or connect().

A typical interruption produces on_close, followed by on_reconnected once recovery succeeds. During deliberate shutdown, on_close runs without a later reconnection. These callbacks observe the connection lifecycle and do not themselves stop the FastStream application.

Keep callbacks short: aio-pika waits for async callbacks during connection processing. Use them to update a flag or signal an asyncio.Event, and let a separate task perform slow work. Avoid blocking I/O or awaiting broker shutdown or connection-dependent operations inside a callback.