1. import redis
  2. import time
  3. import logging
  5. # Configure logging
  6. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  8. class QueueMirrorer:
  9. def __init__(self, source_queue_url, destination_queue_url, redis_host='localhost', redis_port=6379, redis_db=0):
  10. """
  11. Initializes the QueueMirrorer.
  13. Args:
  14. source_queue_url: URL of the source message queue (e.g., RabbitMQ).
  15. destination_queue_url: URL of the destination message queue (e.g., Redis).
  16. redis_host: Hostname of the Redis server.
  17. redis_port: Port of the Redis server.
  18. redis_db: Redis database number.
  19. """
  20. self.source_queue_url = source_queue_url
  21. self.destination_queue_url = destination_queue_url
  22. self.redis_host = redis_host
  23. self.redis_port = redis_port
  24. self.redis_db = redis_db
  25. self.redis_client = redis.Redis(host=self.redis_host, port=self.redis_port, db=self.redis_db)
  27. def mirror_queue(self, polling_interval=5):
  28. """
  29. Mirrors messages from the source queue to the destination queue.
  31. Args:
  32. polling_interval: Time in seconds between polling the source queue.
  33. """
  34. while True:
  35. try:
  36. # Consume messages from the source queue
  37. messages = self.consume_messages()
  39. # Publish messages to the destination queue
  40. if messages:
  41. self.publish_messages(messages)
  42. else:
  43. logging.debug("No messages to mirror.")
  45. except Exception as e:
  46. logging.error(f"Error mirroring queue: {e}")
  48. time.sleep(polling_interval)
  50. def consume_messages(self):
  51. """
  52. Consumes messages from the source queue. This is a placeholder; implement based on
  53. your message queue system (e.g., RabbitMQ, Kafka).
  54. """
  55. # Placeholder for message consumption logic. Replace with your queue library.
  56. # Example using a dummy queue:
  57. # return [f"Message {i}" for i in range(3)]
  59. #Simulate a queue with messages
  60. messages = ["Message 1", "Message 2", "Message 3"]
  62. return messages
  64. def publish_messages(self, messages):
  65. """
  66. Publishes messages to the destination queue.
  67. """
  68. for message in messages:
  69. try:
  70. self.redis_client.lpush(self.destination_queue_url, message) # Use lpush for FIFO
  71. logging.info(f"Published message: {message} to {self.destination_queue_url}")
  72. except Exception as e:
  73. logging.error(f"Error publishing message {message} to {self.destination_queue_url}: {e}")
  74. #Fallback: Retry after a short delay
  75. time.sleep(1) #Retry after 1 second
  79. if __name__ == '__main__':
  80. # Example Usage
  81. source_queue = 'source_queue'
  82. destination_queue = 'staging_queue'
  84. mirrorer = QueueMirrorer(source_queue, destination_queue)
  85. mirrorer.mirror_queue()

Add your comment