Feat: init ping method

faststream/broker/core/usecase.py

@@ -27,7 +27,7 @@
     BrokerMiddleware,
     ConnectionType,
     CustomCallable,
-    MsgType,
+    MsgType
 )
 from faststream.exceptions import NOT_CONNECTED_YET
 from faststream.log.logging import set_logger_fmt
@@ -342,3 +342,8 @@ async def publish(
             publish = partial(m(None).publish_scope, publish)
 
         return await publish(msg, **kwargs)
+
+    @abstractmethod
+    async def ping(self, timeout: int | None) -> bool:
+        """Check connection alive."""
+        raise NotImplementedError()

faststream/rabbit/broker/broker.py

@@ -655,3 +655,13 @@ async def declare_exchange(
         """Declares exchange object in **RabbitMQ**."""
         assert self.declarer, NOT_CONNECTED_YET  # nosec B101
         return await self.declarer.declare_exchange(exchange)
+
+    async def ping(self, timeout: int | None) -> bool:
+        await super().ping(timeout)
+        try:
+            channel = await self._connection.channel()
+            await channel.declare_queue(name='test_queue', durable=True)
+            await channel.close()
+            return True
+        except Exception:
+            return False

tests/brokers/base/connection.py

@@ -46,3 +46,21 @@ async def test_connect_by_url_priority(self, settings):
         await broker.connect(**kwargs)
         assert await self.ping(broker)
         await broker.close()
+
+    @pytest.mark.asyncio()
+    async def test_connection_alive(self, settings):
+        kwargs = self.get_broker_args(settings)
+        broker = self.broker()
+        await broker.connect(**kwargs)
+        alive = await broker.ping(timeout=5)
+        assert alive
+        await broker.close()
+
+    @pytest.mark.asyncio()
+    async def test_connection_not_alive(self, settings):
+        kwargs = self.get_broker_args(settings)
+        broker = self.broker()
+        await broker.connect(**kwargs)
+        await broker.close()
+        alive = await broker.ping(timeout=5)
+        assert not alive

tests/brokers/rabbit/test_connect.py

@@ -58,3 +58,19 @@ async def test_connect_merge_args_and_kwargs_native(self, settings):
         broker = self.broker("fake-url")  # will be ignored
         assert await broker.connect(url=settings.url)
         await broker.close()
+
+    @pytest.mark.asyncio()
+    async def test_connection_alive(self, settings):
+        broker = self.broker("fake-url")  # will be ignored
+        await broker.connect(url=settings.url)
+        alive = await broker.ping(timeout=5)
+        assert alive
+        await broker.close()
+
+    @pytest.mark.asyncio()
+    async def test_connection_not_alive(self, settings):
+        broker = self.broker("fake-url")  # will be ignored
+        await broker.connect(url=settings.url)
+        await broker.close()
+        alive = await broker.ping(timeout=5)
+        assert not alive