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

RED-4653: Re-ordered message processing so that ack occurs after publishing the result, to prevent message loss

Browse Source
This commit is contained in:
Viktor Seifert 2022-07-25 11:51:16 +02:00
parent 9339186b86
commit aff1d06364
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pyinfra/queue/queue_manager.py
View File

@ -68,18 +68,22 @@ class QueueManager(object):
def _create_queue_callback(self, process_message_callback: Callable):
def callback(_channel, frame, properties, body):
self.logger.info("Received message from queue")
self.logger.info(f"Received message from queue with delivery_tag {frame.delivery_tag}")
self.logger.debug(f"Processing {(frame, properties, body)}.")
try:
unpacked_message_body = json.loads(body)
callback_result = process_message_callback(unpacked_message_body)
self._channel.basic_ack(frame.delivery_tag)
self.logger.info("Processed message, publishing result to result-queue")
self._channel.basic_publish("", self._output_queue, json.dumps(callback_result).encode())
self.logger.info(
f"Result published, acknowledging incoming message with delivery_tag {frame.delivery_tag}")
self._channel.basic_ack(frame.delivery_tag)
self.logger.info(f"Message with delivery_tag {frame.delivery_tag} processed")
except Exception as ex:
n_attempts = _get_n_previous_attempts(properties) + 1
self.logger.warning(f"Failed to process message, {n_attempts} attempts, error: {str(ex)}")