added method to remove message

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 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
+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)

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()