58 lines
2.2 KiB
Python
58 lines
2.2 KiB
Python
import pika
|
|
import json
|
|
from typing import Callable, Dict, Any
|
|
import logging
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class RabbitMQConsumer:
|
|
def __init__(self, queue_name: str, host: str = 'localhost', port: int = 5672):
|
|
self.queue_name = queue_name
|
|
self.host = host
|
|
self.port = port
|
|
self.callback = None
|
|
|
|
def set_callback(self, callback: Callable[[Dict[str, Any]], None]) -> None:
|
|
self.callback = callback
|
|
|
|
def start_consuming(self) -> None:
|
|
try:
|
|
connection = pika.BlockingConnection(
|
|
pika.ConnectionParameters(host=self.host, port=self.port)
|
|
)
|
|
channel = connection.channel()
|
|
channel.queue_declare(queue=self.queue_name, durable=True)
|
|
|
|
def callback_wrapper(ch, method, properties, body):
|
|
try:
|
|
message = json.loads(body.decode('utf-8'))
|
|
logger.info(f"Received message from queue '{self.queue_name}': {message}")
|
|
|
|
if self.callback:
|
|
self.callback(message)
|
|
|
|
ch.basic_ack(delivery_tag=method.delivery_tag)
|
|
|
|
except IntegrityError as e:
|
|
# Error de negocio: no tiene sentido reintentar
|
|
logger.warning(f"Business error, discarding message: {e}")
|
|
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
|
|
|
|
except Exception as e:
|
|
# Error transitorio (red, DB caída): sí puede resolverse solo
|
|
logger.error(f"Transient error processing message: {e}")
|
|
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
|
|
|
|
channel.basic_consume(
|
|
queue=self.queue_name,
|
|
on_message_callback=callback_wrapper,
|
|
auto_ack=False
|
|
)
|
|
|
|
logger.info(f"[*] Waiting for messages in queue '{self.queue_name}'. Ctrl+C to exit.")
|
|
channel.start_consuming()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in consumer: {e}")
|
|
raise |