From e205674d31fe2734cc4e3b5bcc33a67055311067 Mon Sep 17 00:00:00 2001
From: Gouri Javed <javeda3740@gmail.com>
Date: Sun, 21 Mar 2021 16:34:16 +0530
Subject: [PATCH] added long polling support  (#506)

* added long pooling support for  sqs broker

* changed the parameter name

* added long polling support ,

introduce new parameter in broker sqs block
parameter for long polling is receive_message_wait_time_seconds

eg :
Q_CLUSTER = {
      'name': 'test-queue',
      'sqs': {
         'aws_region': AWS_S3_REGION_NAME,
         'aws_access_key_id': AWS_ACCESS_KEY_ID,
         'aws_secret_access_key': AWS_SECRET_ACCESS_KEY,
         'receive_message_wait_time_seconds':20
     }
}

* added receive_message_wait_time_seconds parameter in test_broker.py

Co-authored-by: Javed Gouri <you@example.com>
---
 django_q/brokers/aws_sqs.py    | 23 ++++++++++++++++++++---
 django_q/tests/test_brokers.py |  1 +
 2 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/django_q/brokers/aws_sqs.py b/django_q/brokers/aws_sqs.py
index 2bc58c8..ae97c3d 100644
--- a/django_q/brokers/aws_sqs.py
+++ b/django_q/brokers/aws_sqs.py
@@ -26,9 +26,22 @@ class Sqs(Broker):
         # sqs supports max 10 messages in bulk
         if Conf.BULK > 10:
             Conf.BULK = 10
-        tasks = self.queue.receive_messages(
-            MaxNumberOfMessages=Conf.BULK, VisibilityTimeout=Conf.RETRY
-        )
+
+        params = {"MaxNumberOfMessages": Conf.BULK, "VisibilityTimeout": Conf.RETRY}
+
+        # sqs long polling
+        sqs_config = Conf.SQS
+        if "receive_message_wait_time_seconds" in sqs_config:
+            wait_time_second = sqs_config.get("receive_message_wait_time_seconds", 20)
+
+            # validation of parameter
+            if not isinstance(wait_time_second, int):
+                raise ValueError("receive_message_wait_time_seconds should be int")
+            if wait_time_second > 20:
+                raise ValueError("receive_message_wait_time_seconds is invalid. Reason: Must be >= 0 and <= 20")
+            params.update({"WaitTimeSeconds": wait_time_second})
+
+        tasks = self.queue.receive_messages(**params)
         if tasks:
             return [(t.receipt_handle, t.body) for t in tasks]
 
@@ -66,6 +79,10 @@ class Sqs(Broker):
         if "aws_region" in config:
             config["region_name"] = config["aws_region"]
             del config["aws_region"]
+
+        if 'receive_message_wait_time_seconds' in config:
+            del config["receive_message_wait_time_seconds"]
+
         return Session(**config)
 
     def get_queue(self):
diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py
index 0068150..41c385d 100644
--- a/django_q/tests/test_brokers.py
+++ b/django_q/tests/test_brokers.py
@@ -194,6 +194,7 @@ def canceled_sqs(monkeypatch):
             "aws_region": os.getenv("AWS_REGION"),
             "aws_access_key_id": os.getenv("AWS_ACCESS_KEY_ID"),
             "aws_secret_access_key": os.getenv("AWS_SECRET_ACCESS_KEY"),
+            "receive_message_wait_time_seconds": 20
         },
     )
     # check broker