fix: Disabled excessive throttling for BatchSubscriber.

airtai · Dec 29, 2024 · 20e3df8 · 20e3df8
1 parent 2af8277
commit 20e3df8
Showing 1 changed file with 6 additions and 10 deletions.
diff --git a/faststream/confluent/subscriber/usecase.py b/faststream/confluent/subscriber/usecase.py
@@ -335,18 +335,14 @@ def __init__(
 
     async def get_msg(self) -> Optional[tuple["Message", ...]]:
         assert self.consumer, "You should setup subscriber at first."  # nosec B101
-
-        messages = await self.consumer.getmany(
-            timeout=self.polling_interval,
-            max_records=self.max_records,
+        return (
+            await self.consumer.getmany(
+                timeout=self.polling_interval,
+                max_records=self.max_records,
+            )
+            or None
         )
 
-        if not messages:  # TODO: why we are sleeping here?
-            await anyio.sleep(self.polling_interval)
-            return None
-
-        return messages
-
     def get_log_context(
         self,
         message: Optional["StreamMessage[tuple[Message, ...]]"],