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>
This commit is contained in:
Gouri Javed
2021-03-21 12:04:16 +01:00
committed by GitHub
co-authored by Javed Gouri
parent 1869a94118
commit e205674d31
2 changed files with 21 additions and 3 deletions
+20 -3
View File
@@ -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):
+1
View File
@@ -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