94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
import threading
|
|
|
|
import pika
|
|
|
|
from queue_message.CookiesPublisher import QUEUE_HOST, CookiesPublisher, MORNING_DATA_CACHE, MORNING_DATA_CACHE_2
|
|
|
|
credentials = pika.PlainCredentials('appointment', 'ZyuhJZ2xEYWhElhpJjy7YEpZGZwNYJz2fHIu')
|
|
|
|
|
|
class MessageTransporter(threading.Thread):
|
|
|
|
def __init__(self, cookiesPublisher: CookiesPublisher,
|
|
queue_to_listen):
|
|
self.cookiesPublisher = cookiesPublisher
|
|
self.queue_to_listen = queue_to_listen
|
|
|
|
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.cookiesPublisher.message_count()
|
|
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)
|
|
|
|
|
|
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()
|
|
message_transporter = MessageTransporter(cookiesPublisher=cookiesPublisher, queue_to_listen=from_queue)
|
|
message_transporter.set_up_connection()
|
|
message_transporter.listen_to_queue()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
migrate_message_to_queue(from_queue=MORNING_DATA_CACHE_2)
|
|
# cookiesPublisher = CookiesPublisher(queue_name="MORNING_DATA_CACHE_BAK")
|
|
# 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()
|