diff --git a/pyinfra/queue/queue_manager.py b/pyinfra/queue/queue_manager.py index 9e945d5..9644fb1 100644 --- a/pyinfra/queue/queue_manager.py +++ b/pyinfra/queue/queue_manager.py @@ -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)}")