43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
import logging
|
|
from typing import Callable
|
|
|
|
import pika
|
|
from retry import retry
|
|
|
|
from pyinfra.rabbitmq import make_connection, make_channel, declare_queue
|
|
|
|
|
|
class ConsumerError(Exception):
|
|
pass
|
|
|
|
|
|
# @retry(pika.exceptions.AMQPConnectionError, delay=5, jitter=(1, 3))
|
|
def consume(queue_name: str, on_message_callback: Callable):
|
|
|
|
logging.info("Starting mini-queue...")
|
|
connection = make_connection()
|
|
channel = make_channel(connection)
|
|
declare_queue(channel, queue_name)
|
|
logging.info("Starting webserver...")
|
|
|
|
while True:
|
|
try:
|
|
logging.info("Waiting for messages...")
|
|
channel.basic_consume(queue=queue_name, auto_ack=False, on_message_callback=on_message_callback)
|
|
channel.start_consuming()
|
|
|
|
except pika.exceptions.ConnectionClosedByBroker as err:
|
|
logging.critical(f"Caught a channel error: {err}, stopping.")
|
|
break
|
|
|
|
except pika.exceptions.AMQPChannelError as err:
|
|
logging.critical(f"Caught a channel error: {err}, stopping.")
|
|
break
|
|
|
|
except pika.exceptions.AMQPConnectionError as err:
|
|
logging.info("No AMPQ-connection found, retrying...")
|
|
logging.debug(err)
|
|
continue
|
|
|
|
raise ConsumerError(f"Error while consuming {queue_name}.")
|