added long polling support

[Javedgouri]

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
}
}

[Koed00]

Before I can merge this it needs to be covered in the tests. Just adding the new receive_message_wait_time_seconds parameter would probably fix the coverage. Would also be great if you can add this parameter to the docs.
Thanks for your addition!

[Javedgouri]

receive_message_wait_time_seconds in which file i have to add this parameter for coverage.

[Koed00]

Just find the canceled_sqs tests in the tests/test_brokers.py
If you add the parameter in one of the sqs task calls, it should be covered.

Let me know if you are having issues with this.

[Koed00]

Thanks a lot for taking the time to do this.

[NigelSwinson]

Alas I think the patch is broken. I'm trying to use it to upgrade may Mailman/django installation. In get_connection, you have:

    if 'receive_message_wait_time_seconds' in config:
        del config["receive_message_wait_time_seconds"]

I think you are doing this as Session() is only expecting the arguments which create an AWS API session, and this isn't a parameter to that call, rather it's a parameter to receive_message(), which is the subsequent call. So without it, the call to create the Session will fail.

But the way the code is written, the del config[] affects the global Conf.SQS object, which means by the time it gets to the first call to dequeue, the receive_message_wait_time_seconds attribute has been deleted, and therefore the WaitTimeSeconds is never applied to the params. Net result is I think the merge as stated has ZERO effect. :o(

I considered two fixes, and went with the safest, but there may be a better one where you deep copy the config object and pass that to create the Session object, but I'm not a python Guru, so was concerned that the Session(**config) was some kind of call by reference, meaning it may want to add state into that object as a side-effect of the call, so opted for the safer fix.

I'm about to move to test the following in production:

[ec2-user django_q]$ git diff aws_sqs.py
diff --git a/mail/mailman/fixes/django_q/aws_sqs.py b/mail/mailman/fixes/django_q/aws_sqs.py
index 256813b5..4f223734 100644
--- a/mail/mailman/fixes/django_q/aws_sqs.py
+++ b/mail/mailman/fixes/django_q/aws_sqs.py
@@ -1,5 +1,6 @@
 from boto3 import Session
 from botocore.client import ClientError
+from warnings import warn

 from django_q.brokers import Broker
 from django_q.conf import Conf
@@ -40,6 +41,8 @@ class Sqs(Broker):
             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})
+        else:
+            warn("receive_message_wait_time_seconds not defined. short polling will increase AWS costs")

         tasks = self.queue.receive_messages(**params)
         if tasks:
@@ -80,13 +83,24 @@ class Sqs(Broker):
             config["region_name"] = config["aws_region"]
             del config["aws_region"]

+        # The config object, in its entirity is passed to Session to create the AWS config. It won't be expecting our receive_message_wait_time_seconds,
+        # so pull it out if it is set, and put it back afterwards.
+        wait_time_seconds = None
         if 'receive_message_wait_time_seconds' in config:
+            wait_time_seconds = config.get("receive_message_wait_time_seconds", 20)
             del config["receive_message_wait_time_seconds"]

-        return Session(**config)
+        session = Session(**config)
+
+        # Restore the receive_message_wait_time_seconds parameter (if it was set)
+        if (not wait_time_seconds is None):
+            config["receive_message_wait_time_seconds"] = wait_time_seconds
+
+        return session

     def get_queue(self):
         self.sqs = self.connection.resource("sqs")
+        config = Conf.SQS

         try:
             # try to return an existing queue by name. If the queue does not

Ideally this would have been caught by a test.

[NigelSwinson]

The fix seems to have worked. The CPU floor of my ECS Task (Docker Task) has now reduced from 5.4% to 1.9%; and the SQS calls from 2K/min to 3/min. FYI this represented ~$35/month direct SQS costs plus indirect costs (from the EC2 capacity needed to maintain 3.5% CPU of a t2a.medium instance). I think django_q should consider turning on long polling by default.

[Javedgouri]

@NigelSwinson i have created a pull request for this issue