1. import time
  2. import threading
  3. from collections import deque
  5. class MessageQueueNormalizer:
  6. def __init__(self, rate_limit_requests_per_second=10):
  7. self.rate_limit_requests_per_second = rate_limit_requests_per_second
  8. self.request_timestamps = deque() # Store timestamps of recent requests
  9. self.lock = threading.Lock() # Ensure thread-safe access
  11. def normalize_and_rate_limit(self, message):
  12. """
  13. Normalizes a message and applies rate limiting.
  14. """
  15. with self.lock:
  16. now = time.time()
  17. # Remove timestamps older than 1 second
  18. while self.request_timestamps and self.request_timestamps[0] <= now - 1:
  19. self.request_timestamps.popleft()
  21. # Check if rate limit is exceeded
  22. if len(self.request_timestamps) >= self.rate_limit_requests_per_second:
  23. wait_time = 1 / self.rate_limit_requests_per_second
  24. time.sleep(wait_time)
  26. # Add current timestamp to the queue
  27. self.request_timestamps.append(now)
  29. # Perform normalization on the message (replace with your actual normalization logic)
  30. normalized_message = self._normalize_message(message)
  31. return normalized_message
  33. def _normalize_message(self, message):
  34. """
  35. Placeholder for message normalization. Replace with your logic.
  36. """
  37. # Example: Convert message to uppercase
  38. return message.upper()
  40. def clear_rate_limit(self):
  41. """Resets the rate limit counter"""
  42. with self.lock:
  43. self.request_timestamps.clear()

Add your comment