refactor: unify ping method code

faststream/kafka/broker/broker.py

@@ -811,8 +811,10 @@ async def publish_batch(
     @override
     async def ping(self, timeout: Optional[float]) -> bool:
         with move_on_after(timeout) as cancel_scope:
-            return not (
-                cancel_scope.cancel_called
-                or self._producer is None
-                or self._producer._producer._closed
-            )
+            if cancel_scope.cancel_called:
+                return False
+
+            if self._producer is None:
+                return False
+
+            return not self._producer._producer._closed

faststream/nats/broker/broker.py

@@ -921,5 +921,8 @@ async def ping(self, timeout: Optional[float]) -> bool:
         with move_on_after(timeout) as cancel_scope:
             if cancel_scope.cancel_called:
                 return False
-            else:
-                return self._connection.is_connected
+
+            if self._connection is None:
+                return False
+
+            return self._connection.is_connected

faststream/rabbit/broker/broker.py

@@ -662,6 +662,8 @@ async def ping(self, timeout: Optional[float]) -> bool:
         with move_on_after(timeout) as cancel_scope:
             if cancel_scope.cancel_called:
                 return False
-            if not self._connection or self._connection.is_closed:
+
+            if self._connection is None:
                 return False
-        return True
+
+            return not self._connection.is_closed

faststream/redis/broker/broker.py

@@ -484,5 +484,8 @@ async def ping(self, timeout: Optional[float]) -> bool:
         with move_on_after(timeout) as cancel_scope:
             if cancel_scope.cancel_called:
                 return False
-            else:
-                return await self._connection.ping()
+
+            if self._connection is None:
+                return False
+
+            return await self._connection.ping()