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

RED-4653: Added code to cancel the queue subscription on application exit to queue manager so that it can exit gracefully

Browse Source
This commit is contained in:
Viktor Seifert 2022-07-25 15:07:40 +02:00
parent 64d8e0bd15
commit 96e8ac4316
1 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
pyinfra/queue/queue_manager.py
View File

@ -1,9 +1,12 @@
import atexit
import json
import logging
import pika, pika.exceptions
import signal
from typing import Callable
import pika
import pika.exceptions
from pyinfra.config import get_config, Config
CONFIG = get_config()
@ -32,6 +35,10 @@ class QueueManager(object):
def __init__(self, config: Config = CONFIG):
connection_params = get_connection_params(config)
atexit.register(self.cancel_queue_subscription)
signal.signal(signal.SIGTERM, self.cancel_queue_subscription)
signal.signal(signal.SIGINT, self.cancel_queue_subscription)
self._connection = pika.BlockingConnection(parameters=connection_params)
self._channel = self._connection.channel()
self._channel.basic_qos(prefetch_count=1)
@ -63,13 +70,14 @@ class QueueManager(object):
self.cancel_queue_subscription()
def cancel_queue_subscription(self):
if self._consumer_token:
if self._consumer_token and self._connection:
self.logger.info(f"Cancelling subscription for consumer-tag: {self._consumer_token}")
self._channel.basic_cancel(self._consumer_token)
self._connection.close()
self._consumer_token = None
else:
self.logger.warning("No consumer-tag found, a possible subscription cannot be cancelled")
self.logger.info("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):