Compare commits

...

2 Commits

Author SHA1 Message Date
223eeff0be thx gpt 2024-06-12 22:45:54 -04:00
cb64c49092 use env var for rabbitmq host 2024-06-12 22:00:21 -04:00
3 changed files with 76 additions and 21 deletions

View File

@ -1,5 +1,7 @@
import asyncio
import os
import sys
import time
import threading
from queue import Queue
from loguru import logger
@ -14,6 +16,10 @@ 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(
@ -28,15 +34,20 @@ async def main():
monitor_thread.start()
# RabbitMQ
host = 'localhost'
host = os.getenv("RABBITMQ_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})
queue.task_done()
except Exception as e:
logger.error(f"bigtime issue here: {e}")
queue.put(alert_text)
# try:
# await bot.send_markdown(event)

View File

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

View File

@ -1,5 +1,6 @@
import json
import pika
import time
from loguru import logger
@ -10,20 +11,39 @@ 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
while retries < self.max_retries:
try:
self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, port=self.port))
self.channel = self.connection.channel()
self.channel.exchange_declare(exchange=self.exchange, exchange_type=self.exchange_type)
logger.info(f"Successfully connected to RabbitMQ at {self.host}:{self.port}")
return
except Exception as e:
logger.warning(f"Could not connect to RabbitMQ: {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):
if self.connection.is_closed:
retries = 0
while retries < self.max_retries:
if self.connection is None or not self.connection.is_open:
logger.info("Connection is closed. Reconnecting...")
self.connect()
try:
@ -33,9 +53,32 @@ class RabbitMQProducer:
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 not self.connection.is_closed:
if self.connection and self.connection.is_open:
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()