Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

quickfix consumer.consume_and_publish bug

Browse Source
This commit is contained in:
Julius Unverfehrt 2022-03-15 13:29:19 +01:00
parent 0ce30cafbf
commit 8744cdeb6d
3 changed files with 7 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pyinfra/pyampq/consumer.py
View File

@ -7,7 +7,7 @@ class Consumer:
self.callback = callback
def consume_and_publish(self):
yield from self.queue_manager.consume_and_publish(self.callback)
self.queue_manager.consume_and_publish(self.callback)
def consume(self):
yield from self.queue_manager.consume()

10
pyinfra/pyampq/queue_manager/pika_queue_manager.py
View File

@ -74,8 +74,8 @@ class PikaQueueManager(QueueManager):
args = {"x-dead-letter-exchange": "", "x-dead-letter-routing-key": dead_letter_queue}
self.channel.queue_declare(input_queue, arguments=args, auto_delete=False, durable=False)
self.channel.queue_declare(output_queue, arguments=args, auto_delete=False, durable=False)
self.channel.queue_declare(input_queue, arguments=args, auto_delete=False)
self.channel.queue_declare(output_queue, arguments=args, auto_delete=False)
def republish(self, body, n_current_attempts, frame):
self.channel.basic_publish(
@ -122,12 +122,12 @@ class PikaQueueManager(QueueManager):
logger.debug("Consuming")
return self.channel.consume(self._input_queue)
def consume_and_publish(self, callback):
def consume_and_publish(self, visitor):
logger.info(f"Consuming with callback {callback.__name__}")
logger.info(f"Consuming with callback {visitor.callback.__name__}")
for message in self.consume():
self.publish_response(message, callback)
self.publish_response(message, visitor)
def clear(self):
self.channel.queue_purge(self._input_queue)

2
pyinfra/rabbitmq.py
View File

@ -11,7 +11,7 @@ def make_channel(connection) -> pika.adapters.blocking_connection.BlockingChanne
def declare_queue(channel, queue: str):
args = {"x-dead-letter-exchange": "", "x-dead-letter-routing-key": CONFIG.rabbitmq.queues.dead_letter}
return channel.queue_declare(queue=queue, auto_delete=False, arguments=args, durable=True)
return channel.queue_declare(queue=queue, auto_delete=False, arguments=args)
def read_connection_params():