Merge pull request #18 from cuda-networks/test_sqs

Test sqs

Author: alexeyts

eb_sqs/worker/service.py

@@ -56,16 +56,33 @@ def process_queues(self, queue_names):
                     ', '.join([queue.url for queue in queues])
                 ))
 
+            logger.debug('[django-eb-sqs] Processing {} queues'.format(len(queues)))
             self.process_messages(queues, worker)
 
     def process_messages(self, queues, worker):
         # type: (list, Worker) -> None
         for queue in queues:
             try:
                 messages = self.poll_messages(queue)
+                logger.debug('[django-eb-sqs] Polled {} messages'.format(len(messages)))
+
+                msg_entries = []
 
                 for msg in messages:
                     self.process_message(msg, worker)
+                    msg_entries.append({
+                            'Id': msg.message_id,
+                            'ReceiptHandle': msg.receipt_handle
+                    })
+
+                if len(messages) > 0:
+                    response = queue.delete_messages(Entries=msg_entries)
+                    logger.debug('[django-eb-sqs] Deleted {} messages successfully'.format(
+                        len(response.get('Successful', []))
+                    ))
+                    logger.debug('[django-eb-sqs] Failed deleting {} messages'.format(
+                        len(response.get('Failed', []))
+                    ))
             except Exception as exc:
                 logger.warning('[django-eb-sqs] Error polling queue {}: {}'.format(queue.url, exc), exc_info=1)
 
@@ -84,9 +101,6 @@ def process_message(self, msg, worker):
             logger.debug('[django-eb-sqs] Processed message {}'.format(msg.message_id))
         except Exception as exc:
             logger.error('[django-eb-sqs] Unhandled error: {}'.format(exc), exc_info=1)
-        finally:
-            msg.delete()
-            logger.debug('[django-eb-sqs] Deleted message {}'.format(msg.message_id))
 
     def get_queues_by_names(self, sqs, queue_names):
         # type: (ServiceResource, list) -> list