feat: add long polling and batching parameters

Author: mikeshultz

taskiq_sqs/broker.py

@@ -24,6 +24,8 @@ class SQSBroker(AsyncBroker):
     def __init__(
         self,
         sqs_queue_url: str,
+        wait_time_seconds: int = 0,  # Used for long polling
+        max_number_of_messages: int = 1,  # size of batch to receive from the queue
         result_backend: Optional[AsyncResultBackend] = None,
         task_id_generator: Optional[Callable[[], str]] = None,
     ) -> None:
@@ -32,6 +34,9 @@ def __init__(
         self._sqs = boto3.resource("sqs")
         self._sqs_queue: Optional[Queue] = None
 
+        self.wait_time_seconds = wait_time_seconds
+        self.max_number_of_messages = max_number_of_messages
+
     async def _get_queue(self) -> Queue:
         queue_name = self.sqs_queue_url.split("/")[-1]
 
@@ -100,7 +105,13 @@ async def listen(self) -> AsyncGenerator[Union[bytes, AckableMessage], None]:
             last_had_message = False
 
             for message in await asyncify(queue.receive_messages)(
-                MessageAttributeNames=[".*"]
+                MessageAttributeNames=[".*"],
+                # If there's competition on this queue (multiple processes of workers pulling from
+                # the same queue), and processing takes longer than the visibility timeout, multiple
+                # workers may end up processing the same message.
+                MaxNumberOfMessages=self.max_number_of_messages,
+                # Use long poling.
+                WaitTimeSeconds=self.wait_time_seconds,
             ):
                 try:
                     if message.message_attributes: