1. import json
  2. import time
  3. import threading
  4. import logging
  5. from collections import deque
  7. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  9. class PayloadScheduler:
  10. def __init__(self, max_payload_size_kb=1024):
  11. self.payloads = deque() # Use deque for efficient adding/removing from both ends
  12. self.max_payload_size_kb = max_payload_size_kb
  13. self.running = False
  16. def add_payload(self, payload_json, execution_delay_sec):
  17. """Adds a JSON payload and its execution delay to the scheduler."""
  18. try:
  19. payload = json.loads(payload_json)
  20. except json.JSONDecodeError:
  21. logging.error("Invalid JSON payload.")
  22. return False
  24. if self._payload_size_exceeds_limit(payload_json):
  25. logging.warning("Payload size exceeds the limit. Will be skipped.")
  26. return False
  28. self.payloads.append((payload, execution_delay_sec))
  29. logging.info(f"Payload added. Executing in {execution_delay_sec} seconds.")
  30. return True
  33. def _payload_size_exceeds_limit(self, payload_json):
  34. """Checks if the payload size exceeds the defined limit.
  35. This is a basic implementation and can be improved.
  36. """
  37. try:
  38. payload = json.loads(payload_json)
  39. payload_size_kb = len(json.dumps(payload).encode('utf-8')) / 1024
  40. return payload_size_kb > self.max_payload_size_kb
  41. except (json.JSONDecodeError, TypeError):
  42. return True # Assume large if invalid JSON
  45. def _execute_payload(self, payload):
  46. """Executes the given payload."""
  47. try:
  48. logging.info(f"Executing payload: {payload}")
  49. # Replace this with your actual payload execution logic
  50. # This is just a placeholder.
  51. # Example:
  52. # exec(payload) # Use with caution!
  53. time.sleep(2) # Simulate some work
  54. logging.info(f"Payload execution completed: {payload}")
  57. except Exception as e:
  58. logging.error(f"Error executing payload: {e}")
  62. def run(self):
  63. """Starts the scheduler in a separate thread."""
  64. self.running = True
  65. while self.running:
  66. if self.payloads:
  67. payload, delay = self.payloads.popleft()
  68. time.sleep(delay)
  69. threading.Thread(target=self._execute_payload, args=(payload,)).start()
  70. else:
  71. time.sleep(0.1) # Check for new payloads periodically
  74. def stop(self):
  75. """Stops the scheduler."""
  76. self.running = False
  77. logging.info("Scheduler stopped.")
  80. if __name__ == '__main__':
  81. scheduler = PayloadScheduler()
  83. # Example usage:
  84. payload1 = '{"message": "Hello from payload 1"}'
  85. payload2 = '{"number": 123}'
  86. payload3 = '{"data": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]}' #potentially large
  88. scheduler.add_payload(payload1, 1) # Execute after 1 second
  89. scheduler.add_payload(payload2, 3) # Execute after 3 seconds
  90. scheduler.add_payload(payload3, 5) # Execute after 5 seconds
  93. scheduler.run()
  94. time.sleep(10) # Let the scheduler run for a while
  95. scheduler.stop()

Add your comment