1. import redis
  2. import json
  3. import time
  4. import logging
  6. # Configure logging
  7. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  9. class MessageQueueMirror:
  10. def __init__(self, main_queue_url, mirror_queue_url, redis_host='localhost', redis_port=6379):
  11. """
  12. Initializes the MessageQueueMirror.
  14. Args:
  15. main_queue_url (str): URL of the main message queue (e.g., RabbitMQ, Kafka).
  16. mirror_queue_url (str): URL of the mirror message queue (e.g., Redis).
  17. redis_host (str): Redis host.
  18. redis_port (int): Redis port.
  19. """
  20. self.main_queue_url = main_queue_url
  21. self.mirror_queue_url = mirror_queue_url
  22. self.redis_host = redis_host
  23. self.redis_port = redis_port
  24. self.redis_client = redis.Redis(host=self.redis_host, port=self.redis_port, decode_responses=True)
  26. def mirror_messages(self, message_queue_name):
  27. """
  28. Mirrors messages from a specified queue in the main queue to the mirror queue.
  30. Args:
  31. message_queue_name (str): The name of the queue to mirror.
  32. """
  33. try:
  34. # Consume messages from the main queue
  35. main_queue_consumer = self._consume_messages(message_queue_name)
  37. # Push messages to the mirror queue
  38. for message in main_queue_consumer:
  39. self._push_message_to_mirror(message)
  41. main_queue_consumer.close() # Close the consumer
  42. logging.info(f"Mirroring messages from {message_queue_name} completed.")
  44. except Exception as e:
  45. logging.error(f"Error mirroring messages from {message_queue_name}: {e}")
  47. def _consume_messages(self, message_queue_name):
  48. """
  49. Consumes messages from the main queue and returns an iterator.
  51. Args:
  52. message_queue_name (str): The name of the queue to consume from.
  54. Yields:
  55. dict: Each message as a dictionary.
  56. """
  57. # Simulate consuming from a main queue (replace with actual consumer logic)
  58. # In a real-world scenario, this would involve connecting to the main queue
  59. # and using a consumer library (e.g., pika for RabbitMQ, kafka-python for Kafka).
  60. messages = [
  61. {"message_id": 1, "data": "Message 1"},
  62. {"message_id": 2, "data": "Message 2"},
  63. {"message_id": 3, "data": "Message 3"}
  64. ]
  65. for message in messages:
  66. yield message
  69. def _push_message_to_mirror(self, message):
  70. """
  71. Pushes a message to the mirror queue.
  73. Args:
  74. message (dict): The message to push.
  75. """
  76. try:
  77. # Serialize the message to JSON
  78. message_json = json.dumps(message)
  80. # Push the message to the mirror queue
  81. self.redis_client.lpush(self.mirror_queue_url, message_json)
  82. logging.debug(f"Pushed message {message['message_id']} to mirror queue.")
  84. except Exception as e:
  85. logging.error(f"Error pushing message to mirror queue: {e}")
  88. if __name__ == '__main__':
  89. # Example Usage
  90. mirror = MessageQueueMirror("main_queue", "mirror_queue") # Replace with your queue URLs
  91. mirror.mirror_messages("my_queue")

Add your comment