feat(rabbit): add connection event callbacks

docs/docs/en/rabbit/broker-events.md

@@ -0,0 +1,67 @@
+---
+description: >-
+  Observe RabbitMQ connection closure and reconnection with callbacks on
+  FastStream's RabbitBroker.
+search:
+  boost: 10
+---
+
+# 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](https://docs.aio-pika.com/_modules/aio_pika/connection.html){.external-link target="_blank"} and [robust connection implementation](https://docs.aio-pika.com/_modules/aio_pika/robust_connection.html){.external-link target="_blank"} 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.
+
+```python linenums="1" hl_lines="12-23 28-29"
+{!> docs_src/rabbit/broker_events.py !}
+```
+
+With a RabbitMQ server running locally, save this example as `broker_events.py` and run:
+
+```shell
+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.
+
+!!! note "Callback dependencies"
+    aio-pika invokes these callbacks directly. FastStream dependency injection does not apply to them: do not add `Logger`, `Context`, or message parameters to a connection callback. The example uses a standard Python logger for callbacks and an injected `Logger` for the subscriber.
+
+## Callback reference
+
+| Argument         | Signature                                                                                          | Event                                                    |
+|------------------|----------------------------------------------------------------------------------------------------|----------------------------------------------------------|
+| `on_close`       | `async def on_close(connection: AbstractConnection \| None, error: BaseException \| None) -> None` | The connection closes, including during normal shutdown. |
+| `on_reconnected` | `async def on_reconnected(connection: AbstractRobustConnection \| None) -> None`                   | The robust connection reconnects.                        |
+
+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](https://docs.aio-pika.com/_modules/aio_pika/tools.html){.external-link target="_blank"}.
+
+### 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`](../getting-started/middlewares/exception.md){.internal-link} 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](../getting-started/lifespan/hooks.md){.internal-link}.
+
+## 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.

docs/docs/navigation_template.txt

@@ -97,6 +97,7 @@ search:
     - [Queue Arguments](rabbit/queue_arguments.md)
     - [Message Information](rabbit/message.md)
     - [Security Configuration](rabbit/security.md)
+    - [Broker Events](rabbit/broker-events.md)
     - [How-To](howto/rabbit/index.md)
         - [Batch Subscriber](howto/rabbit/batch.md)
 - [MQTT](mqtt/index.md)

docs/docs_src/rabbit/broker_events.py

@@ -0,0 +1,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)

faststream/rabbit/broker/broker.py

@@ -52,7 +52,15 @@
         RobustExchange,
         RobustQueue,
     )
-    from aio_pika.abc import DateType, HeadersType, SSLOptions, TimeoutType
+    from aio_pika.abc import (
+        AbstractConnection,
+        AbstractRobustConnection,
+        CallbackType,
+        DateType,
+        HeadersType,
+        SSLOptions,
+        TimeoutType,
+    )
     from fast_depends.dependencies import Dependant
     from fast_depends.library.serializer import SerializerProto
     from yarl import URL
@@ -89,6 +97,12 @@ def __init__(
         timeout: "TimeoutType" = None,
         fail_fast: bool = True,
         reconnect_interval: "TimeoutType" = 5.0,
+        on_close: Sequence[
+            "CallbackType[AbstractConnection, [BaseException | None], None]"
+        ]
+        | None = None,
+        on_reconnected: Sequence["CallbackType[AbstractRobustConnection, [], None]"]
+        | None = None,
         default_channel: Optional["Channel"] = None,
         app_id: str | None = SERVICE_NAME,
         # broker base args
@@ -129,6 +143,8 @@ def __init__(
             timeout: Connection establishment timeout.
             fail_fast: Broker startup raises `AMQPConnectionError` if RabbitMQ is unreachable.
             reconnect_interval: Time to sleep between reconnection attempts.
+            on_close: Callbacks receiving the connection and closing exception when the connection closes, including during shutdown. Supports sync and async functions. Execution order is not guaranteed.
+            on_reconnected: Callbacks receiving the connection after reconnection. Supports sync and async functions. Execution order is not guaranteed.
             default_channel: Default channel settings to use.
             app_id: Application name to mark outgoing messages by.
             graceful_timeout: Graceful shutdown timeout. Broker waits for all running subscribers completion before shut down.
@@ -238,6 +254,8 @@ def __init__(
         )
 
         self._channel: RobustChannel | None = None
+        self._on_close = tuple(on_close) if on_close is not None else ()
+        self._on_reconnected = tuple(on_reconnected) if on_reconnected is not None else ()
 
     @override
     async def _connect(self) -> "RobustConnection":
@@ -246,6 +264,11 @@ async def _connect(self) -> "RobustConnection":
             await connect_robust(**self._connection_kwargs),
         )
 
+        for close_callback in self._on_close:
+            connection.close_callbacks.add(close_callback)
+        for reconnect_callback in self._on_reconnected:
+            connection.reconnect_callbacks.add(reconnect_callback)
+
         if self._channel is None:
             self.config.connect(connection)
             self._channel = await self.config.channel_manager.get_channel()

tests/brokers/rabbit/test_broker_events.py

@@ -0,0 +1,80 @@
+import asyncio
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from faststream.rabbit import RabbitBroker
+from tests.marks import require_aiopika
+
+if TYPE_CHECKING:
+    from aio_pika.abc import AbstractRobustConnection
+
+
+@pytest.mark.parametrize("callback_type", (MagicMock, AsyncMock))
+@pytest.mark.connected()
+@pytest.mark.asyncio()
+@pytest.mark.rabbit()
+@require_aiopika
+async def test_close_callback_survives_broker_restart(
+    settings: Any,
+    callback_type: type[MagicMock],
+) -> None:
+    on_close = [callback_type(), callback_type()]
+    on_reconnected = (callback_type(), callback_type())
+    broker = RabbitBroker(settings.url, on_close=on_close, on_reconnected=on_reconnected)
+
+    for _ in range(2):
+        async with broker:
+            connection = await broker.connect()
+            assert await broker.connect() is connection
+            for callback in (*on_close, *on_reconnected):
+                callback.assert_not_called()
+
+        for callback in on_close:
+            callback.assert_called_once()
+            assert callback.call_args.args[0] is connection
+            if isinstance(callback, AsyncMock):
+                callback.assert_awaited_once_with(*callback.call_args.args)
+            callback.reset_mock()
+        for callback in on_reconnected:
+            callback.assert_not_called()
+
+
+@pytest.mark.parametrize("callback_type", (MagicMock, AsyncMock))
+@pytest.mark.connected()
+@pytest.mark.asyncio()
+@pytest.mark.rabbit()
+@require_aiopika
+async def test_reconnected_callback_after_automatic_recovery(
+    settings: Any,
+    callback_type: type[MagicMock],
+) -> None:
+    recovered: asyncio.Queue[AbstractRobustConnection | None] = asyncio.Queue()
+    on_close = (MagicMock(), MagicMock())
+    on_reconnected = [
+        callback_type(side_effect=recovered.put_nowait),
+        callback_type(side_effect=recovered.put_nowait),
+    ]
+    broker = RabbitBroker(
+        settings.url,
+        reconnect_interval=0.01,
+        on_close=on_close,
+        on_reconnected=on_reconnected,
+    )
+
+    async with broker:
+        connection = await broker.connect()
+        assert connection.transport is not None
+        await connection.transport.connection.close()
+        for _ in on_reconnected:
+            assert await asyncio.wait_for(recovered.get(), timeout=5) is connection
+        for callback in on_reconnected:
+            if isinstance(callback, AsyncMock):
+                callback.assert_awaited_once_with(connection)
+            else:
+                callback.assert_called_once_with(connection)
+        for callback in on_close:
+            callback.assert_called_once()
+            assert callback.call_args.args[0] is connection
+        assert await broker.ping(timeout=1)

tests/docs/rabbit/test_broker_events.py

@@ -0,0 +1,40 @@
+import logging
+
+import pytest
+
+from docs.docs_src.rabbit.broker_events import (
+    app,
+    broker,
+    handle,
+    on_close,
+    on_reconnected,
+)
+from faststream.rabbit import TestApp, TestRabbitBroker
+from tests.marks import require_aiopika
+
+
+@pytest.mark.asyncio()
+@pytest.mark.rabbit()
+@require_aiopika
+async def test_broker_events_example(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level(logging.INFO, logger="rabbit.events"):
+        await on_close(None, RuntimeError("Connection lost"))
+        await on_reconnected(None)
+        async with TestRabbitBroker(broker), TestApp(app):
+            await broker.publish("hello", "events.example")
+            handle.mock.assert_called_once_with("hello")
+
+    assert "RabbitMQ connection closed: RuntimeError('Connection lost')" in caplog.text
+    assert "RabbitMQ connection restored" in caplog.text
+
+
+@pytest.mark.connected()
+@pytest.mark.asyncio()
+@pytest.mark.rabbit()
+@require_aiopika
+async def test_broker_events_example_shutdown(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level(logging.INFO, logger="rabbit.events"):
+        async with broker:
+            pass
+
+    assert "RabbitMQ connection closed during shutdown" in caplog.text

tests/mypy/rabbit.py

@@ -2,6 +2,7 @@
 
 import prometheus_client
 from aio_pika import IncomingMessage
+from aio_pika.abc import AbstractConnection, AbstractRobustConnection
 from aiormq.abc import ConfirmationFrameType
 from typing_extensions import assert_type
 
@@ -20,6 +21,31 @@
 from faststream.rabbit.subscriber.usecase import RabbitSubscriber
 
 
+def sync_on_close(
+    connection: AbstractConnection | None, error: BaseException | None
+) -> None: ...
+
+
+async def async_on_close(
+    connection: AbstractConnection | None, error: BaseException | None
+) -> None: ...
+
+
+def sync_on_reconnected(connection: AbstractRobustConnection | None) -> None: ...
+
+
+async def async_on_reconnected(connection: AbstractRobustConnection | None) -> None: ...
+
+
+RabbitBroker(on_close=[sync_on_close], on_reconnected=[sync_on_reconnected])
+RabbitBroker(on_close=(async_on_close,), on_reconnected=(async_on_reconnected,))
+RabbitBroker(
+    on_close=[sync_on_close, async_on_close],
+    on_reconnected=[sync_on_reconnected, async_on_reconnected],
+)
+RabbitBroker(on_close=[], on_reconnected=())
+
+
 async def check_multiple_test_brokers() -> None:
     async with TestRabbitBroker(RabbitBroker()) as br1:
         await br1.publish(None, "test")