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

RED-4653: Moved queue cancellation to a separate method so that it can be called on application exit

Browse Source
This commit is contained in:
Viktor Seifert 2022-07-25 13:57:06 +02:00
parent aff1d06364
commit 64d8e0bd15
1 changed files with 13 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
pyinfra/queue/queue_manager.py
View File

@ -53,18 +53,23 @@ class QueueManager(object):
callback = self._create_queue_callback(process_message_callback)
self.logger.info("Consuming from queue")
consumer_tag = None
self._consumer_token = None
try:
consumer_tag = self._channel.basic_consume(self._input_queue, callback)
self.logger.info(f"Registered with consumer-tag: {consumer_tag}")
self._consumer_token = self._channel.basic_consume(self._input_queue, callback)
self.logger.info(f"Registered with consumer-tag: {self._consumer_token}")
self._channel.start_consuming()
finally:
self.logger.warning("An unhandled exception occurred while consuming messages. Consuming will stop.")
if consumer_tag:
self.logger.info(f"Cancelling subscription for consumer-tag: {consumer_tag}")
self._channel.basic_cancel(consumer_tag)
else:
self.logger.warning("No consumer-tag found, a possible subscription cannot be cancelled")
self.cancel_queue_subscription()
def cancel_queue_subscription(self):
if self._consumer_token:
self.logger.info(f"Cancelling subscription for consumer-tag: {self._consumer_token}")
self._channel.basic_cancel(self._consumer_token)
self._consumer_token = None
else:
self.logger.warning("No consumer-tag found, a possible subscription cannot be cancelled")
def _create_queue_callback(self, process_message_callback: Callable):
def callback(_channel, frame, properties, body):