inactivity_timeout for channel.consume

pyinfra/queue/consumer.py

@@ -9,5 +9,5 @@ class Consumer:
     def consume_and_publish(self):
         self.queue_manager.consume_and_publish(self.callback)
 
-    def consume(self):
+    def consume(self, **kwargs):
-        yield from self.queue_manager.consume()
+        yield from self.queue_manager.consume(**kwargs)

pyinfra/queue/queue_manager/pika_queue_manager.py

@@ -117,9 +117,9 @@ class PikaQueueManager(QueueManager):
     def pull_request(self):
         return self.channel.basic_get(self._input_queue)
 
-    def consume(self):
+    def consume(self, inactivity_timeout=None):
         logger.debug("Consuming")
-        return self.channel.consume(self._input_queue)
+        return self.channel.consume(self._input_queue, inactivity_timeout=inactivity_timeout)
 
     def consume_and_publish(self, visitor):
 

pyinfra/queue/queue_manager/queue_manager.py

@@ -27,7 +27,7 @@ class QueueManager(abc.ABC):
         pass
 
     @abc.abstractmethod
-    def consume(self):
+    def consume(self, **kwargs):
         pass
 
     @abc.abstractmethod

pyinfra/test/queue/queue_manager_mock.py

@@ -22,7 +22,7 @@ class QueueManagerMock(QueueManager):
     def pull_request(self):
         return self._input_queue.popleft()
 
-    def consume(self):
+    def consume(self, **kwargs):
         while self._input_queue:
             yield self.pull_request()
 

pyinfra/test/unit_tests/consumer_test.py

@@ -85,7 +85,7 @@ class TestConsumer:
             storage.put_object(**get_object_descriptor(message), data=gzip.compress(data))
             queue_manager.publish_request(message)
 
-        requests = consumer.consume()
+        requests = consumer.consume(inactivity_timeout=5)
 
         for _, r in zip(items, requests):
             queue_manager.publish_response(r, visitor)