1. import logging
  2. import time
  3. import traceback
  4. import queue
  6. # Configure logging
  7. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  10. class MessageQueueNormalizer:
  11. def __init__(self, queue_url, normalization_function, batch_size=100):
  12. self.queue_url = queue_url
  13. self.normalization_function = normalization_function # Function to normalize data
  14. self.queue = queue.Queue()
  15. self.running = True
  17. def enqueue_messages(self, messages):
  18. """Enqueues messages into the queue."""
  19. for message in messages:
  20. self.queue.put(message)
  22. def normalize_messages(self):
  23. """Normalizes messages from the queue."""
  24. while self.running:
  25. try:
  26. messages = self.queue.get(timeout=5) # Wait up to 5 seconds for messages
  27. if messages is None:
  28. # Signal to stop processing
  29. break
  31. normalized_messages = self.normalization_function(messages)
  32. # Process normalized messages (e.g., write to database)
  33. # print(f"Normalized messages: {normalized_messages}") #for debugging
  34. self.queue.task_done() # Mark task as complete
  36. except queue.Empty:
  37. # Queue is empty, continue to check for more messages
  38. pass
  39. except Exception as e:
  40. logging.error(f"Error normalizing messages: {e}")
  41. traceback.print_exc() # Print full traceback
  42. time.sleep(10) # Wait before retrying
  44. def start(self):
  45. """Starts the normalization process."""
  46. self.running = True
  47. logging.info("Normalization process started.")
  48. self.normalize_messages()
  50. def stop(self):
  51. """Stops the normalization process gracefully."""
  52. logging.info("Stopping normalization process...")
  53. self.running = False
  54. # Signal the worker to stop by putting None in the queue
  55. for _ in range(10): #put None multiple times to make sure the worker receives it
  56. self.queue.put(None)
  57. self.queue.join() #wait for all tasks to be completed
  58. logging.info("Normalization process stopped.")
  61. if __name__ == '__main__':
  62. # Example Usage
  63. def example_normalization(messages):
  64. """Example normalization function."""
  65. normalized = []
  66. for msg in messages:
  67. normalized.append(msg.upper()) # Convert to uppercase
  68. return normalized
  70. # Create a queue URL (replace with your actual queue URL)
  71. queue_url = "your_queue_url"
  73. # Create a MessageQueueNormalizer instance
  74. normalizer = MessageQueueNormalizer(queue_url, example_normalization)
  76. # Sample messages
  77. sample_messages = ["hello", "world", "python"]
  79. # Enqueue messages
  80. normalizer.enqueue_messages(sample_messages)
  82. # Start normalization
  83. normalizer.start()
  85. # Let it run for a while
  86. time.sleep(20)
  88. # Stop normalization gracefully
  89. normalizer.stop()

Add your comment