diff --git a/workers/MessagerTransporter.py b/workers/MessagerTransporter.py index 7f4624c..b824136 100644 --- a/workers/MessagerTransporter.py +++ b/workers/MessagerTransporter.py @@ -2,7 +2,7 @@ import threading import pika -from queue_message.CookiesPublisher import QUEUE_HOST, CookiesPublisher, MORNING_DATA_CACHE +from queue_message.CookiesPublisher import QUEUE_HOST, CookiesPublisher, MORNING_DATA_CACHE, MORNING_DATA_CACHE_2 credentials = pika.PlainCredentials('appointment', 'ZyuhJZ2xEYWhElhpJjy7YEpZGZwNYJz2fHIu') @@ -24,6 +24,9 @@ class MessageTransporter(threading.Thread): self.channel.basic_consume(queue=self.queue_to_listen, auto_ack=False, on_message_callback=self.on_message) self.channel.start_consuming() + def message_count(self): + return self.channel.queue_declare(queue=self.queue_to_listen, durable=True).method.message_count + def on_message(self, ch, method, properties, body): print(f" [x] Received {body}") _received_object = body.decode("UTF-8") @@ -31,6 +34,9 @@ class MessageTransporter(threading.Thread): print("message count in queue is {}".format(_message_in_queue_count)) self.cookiesPublisher.publish_body(_received_object) ch.basic_ack(delivery_tag=method.delivery_tag) + if self.message_count() == 0: + print("all messages are processed") + exit(0) if __name__ == '__main__': @@ -38,4 +44,4 @@ if __name__ == '__main__': cookiesPublisher.set_up_connection() message_transporter = MessageTransporter(cookiesPublisher=cookiesPublisher, queue_to_listen=MORNING_DATA_CACHE) message_transporter.set_up_connection() - message_transporter.listen_to_queue() \ No newline at end of file + message_transporter.listen_to_queue()