src/main.py

@@ -1,7 +1,5 @@
 import asyncio
-import os
 import sys
-import time
 import threading
 from queue import Queue
 from loguru import logger
@@ -16,10 +14,6 @@ def init_logger():
 
 
 async def main():
-    # LAZY HACK THIS IS BAD
-    # TODO: Get rid of this trash and add retries if RabbitMQ isn't ready
-    time.sleep(5)
-
     config = load_config("config.json")
     queue = Queue()
     monitor = EventMonitor(
@@ -34,20 +28,15 @@ async def main():
     monitor_thread.start()
 
     # RabbitMQ
-    host = os.getenv("RABBITMQ_HOST", "localhost")
+    host = 'localhost'
     port = 5672
     exchange = 'matrix_alerts'
     producer = RabbitMQProducer(host=host, port=port, exchange=exchange)
 
     while True:
         alert_text = queue.get()
-        try:
+        producer.send_event(message={'body': alert_text})
-            producer.send_event(message={'body': alert_text})
+        queue.task_done()
-            queue.task_done()
-        except Exception as e:
-            logger.error(f"bigtime issue here: {e}")
-            queue.put(alert_text)
-
 
         # try:
         #     await bot.send_markdown(event)

src/requirements.txt

@@ -1,4 +1,3 @@
-pika
 loguru
 web3
 requests

src/rmq.py

@@ -1,6 +1,5 @@
 import json
 import pika
-import time
 from loguru import logger
 
 
@@ -11,74 +10,32 @@ class RabbitMQProducer:
         self.exchange = exchange
         self.exchange_type = exchange_type
 
-        self.max_retries = 5
-        self.initial_delay = 1  # Initial delay in seconds for exponential backoff
-
         self.connection = None
         self.channel = None
         self.connect()
 
     def connect(self):
-        retries = 0
+        try:
-        while retries < self.max_retries:
+            self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, port=self.port))
-            try:
+            self.channel = self.connection.channel()
-                self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, port=self.port))
+            self.channel.exchange_declare(exchange=self.exchange, exchange_type=self.exchange_type)
-                self.channel = self.connection.channel()
+        except Exception as e:
-                self.channel.exchange_declare(exchange=self.exchange, exchange_type=self.exchange_type)
+            logger.warning(f"Could not connect to RabbitMQ: {e}")
-                logger.info(f"Successfully connected to RabbitMQ at {self.host}:{self.port}")
-                return
-            except Exception as e:
-                retries += 1
-                wait_time = self.initial_delay * (2 ** (retries - 1))
-                logger.warning(f"Could not connect to RabbitMQ (attempt {retries}/{self.max_retries}): {e}")
-                if retries < self.max_retries:
-                    logger.info(f"Retrying in {wait_time} seconds...")
-                    time.sleep(wait_time)
-                else:
-                    logger.error("All connection attempts failed.")
-                    raise
 
     def send_event(self, message: dict):
-        retries = 0
+        if self.connection.is_closed:
+            self.connect()
 
-        while retries < self.max_retries:
+        try:
-            if self.connection is None or not self.connection.is_open:
+            self.channel.basic_publish(
-                logger.info("Connection is closed. Reconnecting...")
+                exchange=self.exchange,
-                self.connect()
+                routing_key='',  # Empty routing key for fanout exchange
-
+                body=json.dumps(message),
-            try:
+                properties=pika.BasicProperties(content_type='application/json')
-                self.channel.basic_publish(
+            )
-                    exchange=self.exchange,
+        except Exception as e:
-                    routing_key='',  # Empty routing key for fanout exchange
+            logger.warning(f"Failed to send message: {e}")
-                    body=json.dumps(message),
-                    properties=pika.BasicProperties(content_type='application/json')
-                )
-                logger.info(f"Message sent: {message}")
-                return
-            except pika.exceptions.AMQPConnectionError as e:
-                retries += 1
-                wait_time = self.initial_delay * (2 ** (retries - 1))
-                logger.warning(f"Connection error (attempt {retries}/{self.max_retries}): {e}")
-                if retries < self.max_retries:
-                    logger.info(f"Retrying in {wait_time} seconds...")
-                    self.connect()
-                    time.sleep(wait_time)
-                else:
-                    logger.error("All retry attempts failed.")
-                    raise
-            except Exception as e:
-                logger.warning(f"Failed to send message: {e}")
-                raise
 
     def close_connection(self):
-        if self.connection and self.connection.is_open:
+        if self.connection and not self.connection.is_closed:
             self.connection.close()
-
-
-# Usage example:
-if __name__ == "__main__":
-    producer = RabbitMQProducer(host='localhost', port=5672, exchange='logs')
-    try:
-        producer.send_event({"message": "Hello, RabbitMQ!"})
-    finally:
-        producer.close_connection()