Stanislav Hejny
105 lines
5.0 KiB
Python
105 lines
5.0 KiB
Python
import logging
|
|
import aio_pika
|
|
|
|
from amqp.adapter.amq_message_factory import AMQMessageFactory
|
|
from amqp.config.amq_configuration import AMQConfiguration
|
|
from amqp.model.model import AMQMessage, AMQRoute
|
|
from amqp.router.router_producer import RouterProducer
|
|
|
|
|
|
class RabbitMQProducer:
|
|
PREFETCH_COUNT = 1
|
|
|
|
def __init__(self, amq_configuration: AMQConfiguration, router: RouterProducer = None):
|
|
self.channel: aio_pika.RobustChannel | None = None
|
|
self.connection: aio_pika.RobustConnection | None = None
|
|
self.rpc_exchange: aio_pika.RobustExchange | None = None
|
|
self.reply_exchange: aio_pika.RobustExchange | None = None
|
|
self.amq_configuration = amq_configuration
|
|
logging.info("*******************************************")
|
|
logging.info("******** RabbitMQProducer.init(): %s",
|
|
self.amq_configuration.dispatch.amq_host)
|
|
logging.info("*******************************************")
|
|
self.router = router or RouterProducer(amq_configuration)
|
|
# this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
|
|
|
|
async def init(self, loop):
|
|
await self.setup_amq_channel(loop)
|
|
|
|
def cleanup(self):
|
|
if self.connection:
|
|
try:
|
|
self.connection.close()
|
|
except Exception as e:
|
|
logging.error("RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
|
|
|
|
async def setup_amq_channel(self, loop):
|
|
"""
|
|
Set up the RabbitMQ connection
|
|
:return: nothing, it applies changes to instance variables
|
|
"""
|
|
self.connection = await aio_pika.connect_robust(
|
|
host=self.amq_configuration.dispatch.amq_host,
|
|
port=self.amq_configuration.dispatch.amq_port,
|
|
login=self.amq_configuration.dispatch.rabbit_mq_user,
|
|
password=self.amq_configuration.dispatch.rabbit_mq_password,
|
|
loop=loop
|
|
)
|
|
logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_closed)
|
|
self.channel = await self.connection.channel()
|
|
await self.channel.set_qos(prefetch_count=self.PREFETCH_COUNT)
|
|
await self.router.init_routing_paths(self.channel)
|
|
self.router.set_route_added_notifier(
|
|
lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s",
|
|
route.exchange))
|
|
|
|
async def send_message(self, message: AMQMessage):
|
|
"""
|
|
The API routine that sends a datagram message (AMQMessage) to the destination. The Destination
|
|
is determined from the AMQMessage metadata section, using `domain` and `path` as key.
|
|
:param message: message to send
|
|
:return: nothing (void) as submitting message to RabbitMQ exchange does not return anything
|
|
meaningful. There will be asynch ACK later that will confirm the message was sent to RMQ.
|
|
"""
|
|
try:
|
|
route = self.router.find_route_by_message(message)
|
|
if route and self.rpc_exchange:
|
|
if route.timeout <= 0:
|
|
# No response expected
|
|
pika_message: aio_pika.message.Message = aio_pika.message.Message(
|
|
body=AMQMessageFactory.serialize(message),
|
|
content_encoding='utf-8',
|
|
delivery_mode=2
|
|
)
|
|
await self.rpc_exchange.publish(
|
|
message=pika_message,
|
|
routing_key=self.router.get_routing_key(message), # context path is a routing key
|
|
)
|
|
logging.info(f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}")
|
|
else:
|
|
# This is RPC call, there should be response
|
|
pika_message: aio_pika.message.Message = aio_pika.message.Message(
|
|
body=AMQMessageFactory.serialize(message),
|
|
content_encoding='utf-8',
|
|
delivery_mode=2,
|
|
correlation_id=str(message.id),
|
|
reply_to=self.router.get_reply_to_queue_name()
|
|
)
|
|
await self.rpc_exchange.publish(
|
|
message=pika_message,
|
|
routing_key=self.router.get_routing_key(message), # context path is a routing key
|
|
)
|
|
logging.info(" [x] basicPublish (RPC call), messageId: %s, to: %s",
|
|
message.id.as_string(), route.as_string())
|
|
else:
|
|
logging.warning("Message not sent, no route for %s/%s", message.domain, message.path)
|
|
|
|
except Exception as err:
|
|
logging.error("Send message failed: %s", err)
|
|
|
|
async def request_routes(self):
|
|
await self.router.request_routes_from_adapters()
|
|
|
|
def find_route(self, domain: str, path: str) -> AMQRoute:
|
|
return self.router.find_route(domain, path)
|