quit programme after all messages processed
This commit is contained in:
@@ -2,7 +2,7 @@ import threading
|
|||||||
|
|
||||||
import pika
|
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')
|
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.basic_consume(queue=self.queue_to_listen, auto_ack=False, on_message_callback=self.on_message)
|
||||||
self.channel.start_consuming()
|
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):
|
def on_message(self, ch, method, properties, body):
|
||||||
print(f" [x] Received {body}")
|
print(f" [x] Received {body}")
|
||||||
_received_object = body.decode("UTF-8")
|
_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))
|
print("message count in queue is {}".format(_message_in_queue_count))
|
||||||
self.cookiesPublisher.publish_body(_received_object)
|
self.cookiesPublisher.publish_body(_received_object)
|
||||||
ch.basic_ack(delivery_tag=method.delivery_tag)
|
ch.basic_ack(delivery_tag=method.delivery_tag)
|
||||||
|
if self.message_count() == 0:
|
||||||
|
print("all messages are processed")
|
||||||
|
exit(0)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
@@ -38,4 +44,4 @@ if __name__ == '__main__':
|
|||||||
cookiesPublisher.set_up_connection()
|
cookiesPublisher.set_up_connection()
|
||||||
message_transporter = MessageTransporter(cookiesPublisher=cookiesPublisher, queue_to_listen=MORNING_DATA_CACHE)
|
message_transporter = MessageTransporter(cookiesPublisher=cookiesPublisher, queue_to_listen=MORNING_DATA_CACHE)
|
||||||
message_transporter.set_up_connection()
|
message_transporter.set_up_connection()
|
||||||
message_transporter.listen_to_queue()
|
message_transporter.listen_to_queue()
|
||||||
|
|||||||
Reference in New Issue
Block a user