1. import redis
  2. import time
  3. import logging
  4. import traceback
  5. import queue
  7. # Configure logging
  8. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  10. class QueueMirror:
  11. def __init__(self, source_queue_url, destination_queue_url, redis_host='localhost', redis_port=6379):
  12. self.source_queue_url = source_queue_url
  13. self.destination_queue_url = destination_queue_url
  14. self.redis_host = redis_host
  15. self.redis_port = redis_port
  16. self.source_queue = queue.Queue()
  17. self.redis_client = redis.Redis(host=self.redis_host, port=self.redis_port)
  19. def start_mirroring(self):
  20. try:
  21. # Start consuming from the source queue
  22. self.consume_source()
  23. except Exception as e:
  24. logging.error(f"Error starting mirroring: {e}")
  25. traceback.print_exc() # Print the stack trace for debugging
  28. def consume_source(self):
  29. while True:
  30. try:
  31. message = self.source_queue.get(timeout=1) # Wait up to 1 second for a message
  32. if message is None: # Sentinel value to signal shutdown
  33. break
  34. self.mirror_message(message)
  35. self.source_queue.task_done() # Indicate that a formerly enqueued task is complete
  36. except queue.Empty:
  37. # Queue is empty, continue to the next iteration
  38. pass
  39. except Exception as e:
  40. logging.error(f"Error consuming message: {e}")
  41. traceback.print_exc() # Print stack trace
  43. def mirror_message(self, message):
  44. try:
  45. self.redis_client.rpush(self.destination_queue_url, message)
  46. logging.info(f"Mirrored message: {message}")
  47. except redis.exceptions.ConnectionError as e:
  48. logging.error(f"Redis connection error: {e}")
  49. # Handle connection error, e.g., retry or log and continue
  50. except Exception as e:
  51. logging.error(f"Error mirroring message: {e}")
  52. traceback.print_exc() #Print stack trace
  54. def enqueue_message(self, message):
  55. self.source_queue.put(message)
  57. def stop(self):
  58. # Signal the consumer to stop
  59. self.enqueue_message(None)
  60. self.source_queue.join() # Wait for all tasks to be processed
  61. self.redis_client.close()
  62. logging.info("Mirroring stopped.")
  66. if __name__ == '__main__':
  67. # Example usage (replace with your actual queue URLs)
  68. source_queue_url = 'my_source_queue' # e.g., 'sqs://...' or 'kafka://...'
  69. destination_queue_url = 'my_destination_queue' #e.g., 'redis://...'
  71. mirror = QueueMirror(source_queue_url, destination_queue_url)
  72. mirror.start_mirroring()
  74. try:
  75. # Simulate message production
  76. for i in range(10):
  77. message = f"Message {i}"
  78. mirror.enqueue_message(message)
  79. time.sleep(0.5)
  80. except KeyboardInterrupt:
  81. logging.info("Stopping mirroring...")
  82. finally:
  83. mirror.stop()

Add your comment