1. import pika
  2. import json
  3. import time
  4. import logging
  5. import configparser
  7. class QueueRetry:
  8. def __init__(self, config_file):
  9. self.config = configparser.ConfigParser()
  10. self.config.read(config_file)
  11. self.connection_factory = self.config['rabbitmq']['connection_factory']
  12. self.exchange = self.config['rabbitmq']['exchange']
  13. self.queue = self.config['rabbitmq']['queue']
  14. self.retry_count = self.config['rabbitmq']['retry_count']
  15. self.retry_delay = self.config['rabbitmq']['retry_delay']
  16. self.logging_level = logging.getLogger().setLevel(self.config['logging']['level'])
  17. self.channel = None
  19. def connect(self):
  20. try:
  21. connection = pika.BlockingConnection(self.connection_factory)
  22. self.channel = connection.channel()
  23. self.channel.queue_exists(self.queue) # Check if the queue exists
  24. self.channel.declare_exchange(self.exchange, exchange_type='direct')
  25. self.channel.queue_bind(self.queue, self.exchange, routing_key='my_routing_key') # Replace 'my_routing_key' with your routing key
  26. self.channel.prefetch(1) # Set prefetch count
  27. self.logging_level.info("Connected to RabbitMQ")
  28. return connection
  29. except pika.exceptions.ConnectionRefusedError:
  30. self.logging_level.error("RabbitMQ connection refused.")
  31. return None
  32. except Exception as e:
  33. self.logging_level.error(f"Error connecting to RabbitMQ: {e}")
  34. return None
  37. def publish(self, message):
  38. connection = self.connect()
  39. if not connection:
  40. return False
  42. try:
  43. channel = connection.channel()
  44. channel.basic_publish(exchange=self.exchange,
  45. routing_key='my_routing_key', # Replace 'my_routing_key' with your routing key
  46. body=json.dumps(message))
  47. channel.close()
  48. connection.close()
  49. self.logging_level.info(f"Published message: {message}")
  50. return True
  52. except Exception as e:
  53. self.logging_level.error(f"Error publishing message: {e}")
  54. return False
  56. def consume(self, callback):
  57. connection = self.connect()
  58. if not connection:
  59. return
  61. try:
  62. channel = connection.channel()
  63. channel.basic_consume(self.queue, callback=callback, auto_ack=False)
  64. self.logging_level.info(f"Consuming from queue: {self.queue}")
  65. channel.start_consuming()
  66. except Exception as e:
  67. self.logging_level.error(f"Error consuming message: {e}")
  68. finally:
  69. if channel:
  70. channel.close()
  71. if connection:
  72. connection.close()
  75. def retry_operation(self, operation, max_retries=self.retry_count):
  76. retries = 0
  77. while retries < max_retries:
  78. try:
  79. return operation() # Execute the operation
  80. except Exception as e:
  81. retries += 1
  82. if retries < max_retries:
  83. self.logging_level.warning(f"Operation failed. Retrying in {self.retry_delay} seconds. Attempt {retries}/{max_retries}. Error: {e}")
  84. time.sleep(self.retry_delay)
  85. else:
  86. self.logging_level.error(f"Operation failed after {max_retries} retries. Error: {e}")
  87. raise # Re-raise the exception after all retries have failed
  90. if __name__ == '__main__':
  91. # Example Usage
  92. config_file = 'config.ini' # Replace with your config file
  93. retry_manager = QueueRetry(config_file)
  95. def my_operation():
  96. # Simulate an operation that might fail
  97. print("Attempting operation...")
  98. #raise Exception("Simulated error")
  99. return "Operation successful"
  101. try:
  102. result = retry_manager.retry_operation(my_operation)
  103. print(f"Operation result: {result}")
  104. except Exception as e:
  105. print(f"Operation failed permanently

Add your comment