diff --git a/migrate_to_data_cache_2.py b/migrate_to_data_cache_2.py index 51e29f1..6ee4549 100644 --- a/migrate_to_data_cache_2.py +++ b/migrate_to_data_cache_2.py @@ -1,6 +1,8 @@ - -from queue_message.CookiesPublisher import MORNING_DATA_CACHE_2, MORNING_DATA_CACHE, MORNING_DATA_CACHE_BAK -from workers.MessagerTransporter import migrate_message_to_queue +from queue_message.CookiesPublisher import MORNING_DATA_CACHE_2, MORNING_DATA_CACHE, MORNING_DATA_CACHE_BAK, \ + REQUEST_DATA_QUEUE_DE +from workers.MessagerTransporter import migrate_message_to_queue, empty_message_just_to if __name__ == '__main__': - migrate_message_to_queue(from_queue=MORNING_DATA_CACHE,to_queue=MORNING_DATA_CACHE_2) + migrate_message_to_queue(from_queue=MORNING_DATA_CACHE_BAK, to_queue=MORNING_DATA_CACHE) + # empty_message_just_to(2000, queue_name=MORNING_DATA_CACHE_BAK) + # migrate_message_to_queue(from_queue=MORNING_DATA_CACHE_BAK,to_queue=REQUEST_DATA_QUEUE_DE) diff --git a/workers/MessagerTransporter.py b/workers/MessagerTransporter.py index 8292136..ff54aa0 100644 --- a/workers/MessagerTransporter.py +++ b/workers/MessagerTransporter.py @@ -39,6 +39,43 @@ class MessageTransporter(threading.Thread): exit(0) +class MessageRemover(threading.Thread): + + def __init__(self, limit: int, + queue_to_listen): + self.queue_to_listen = queue_to_listen + self.limit = limit + + def set_up_connection(self): + self.connection = pika.BlockingConnection( + pika.ConnectionParameters(host=QUEUE_HOST, port=5672, credentials=credentials)) + self.channel = self.connection.channel() + + def listen_to_queue(self): + self.channel.basic_qos(prefetch_count=1) + 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") + # _message_in_queue_count = self.message_count + # print("message count in queue is {}".format(_message_in_queue_count)) + ch.basic_ack(delivery_tag=method.delivery_tag) + if self.message_count() <= self.limit: + print("all messages are processed") + exit(0) + + +def empty_message_just_to(left_message_number, queue_name): + _remover = MessageRemover(left_message_number, queue_name) + _remover.set_up_connection() + _remover.listen_to_queue() + + def migrate_message_to_queue(from_queue, to_queue="MORNING_DATA_CACHE_BAK"): cookiesPublisher = CookiesPublisher(queue_name=to_queue) cookiesPublisher.set_up_connection()