177 lines
9.5 KiB
Python
177 lines
9.5 KiB
Python
import logging
|
|
|
|
import aio_pika
|
|
from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
|
|
from pika.exchange_type import ExchangeType
|
|
|
|
from amqp.adapter.data_message_factory import DataMessageFactory
|
|
from amqp.adapter.logging_utils import logging_error
|
|
from amqp.model.model import DataMessage, AMQRoute
|
|
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE, DLQ_EXCHANGE, AMQ_RPC_EXCHANGE
|
|
from amqp.router.router_consumer import RouterConsumer
|
|
from amqp.router.utils import sanitize_as_rabbitmq_name
|
|
|
|
|
|
class RabbitMQClient:
|
|
PREFETCH_COUNT = 1
|
|
DLQ_ARGS = {"x-dead-letter-exchange": DLQ_EXCHANGE}
|
|
|
|
def __init__(self, configuration):
|
|
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 = configuration
|
|
logging.info("*******************************************")
|
|
logging.info("******** RabbitMQProducer.init(): %s",
|
|
self.amq_configuration.dispatch.amq_host)
|
|
logging.info("*******************************************")
|
|
self.router = RouterConsumer(configuration)
|
|
# this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
|
|
|
|
|
|
self.reply_to_exchange: AbstractRobustExchange | None = None # when sending response to RPC request
|
|
self.reply_to_queue: AbstractRobustQueue | None= None # when receiving response to RPC request
|
|
|
|
async def init(self, loop) -> AbstractRobustExchange:
|
|
await self.setup_amq_channel(loop)
|
|
await self.router.init_routing_paths_consumer(self.router.channel)
|
|
cm_reply_to_queue = self.router.get_reply_to_queue_name()
|
|
self.reply_to_queue = await self.channel.declare_queue(
|
|
name=cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False
|
|
)
|
|
self.reply_to_exchange = await self.channel.declare_exchange(
|
|
AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.direct
|
|
)
|
|
await self.reply_to_queue.bind(exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue)
|
|
return self.reply_to_exchange
|
|
|
|
def cleanup(self):
|
|
logging.info(" [*] RabbitMQClient.cleanup()")
|
|
self.router.cleanup() # de-register route(s)
|
|
if self.router.is_open(self.channel):
|
|
try:
|
|
self.channel.close()
|
|
if not self.connection.is_closed:
|
|
self.connection.close()
|
|
except Exception as e:
|
|
logging_error(" [E] RabbitMQClient.PreDestroy cleanup - connection close error: %s", e)
|
|
|
|
async def register_inbound_routes(self, route_mapping: str, callback):
|
|
"""
|
|
Register inbound routes for the AMQ adapter. The route mapping is a comma-separated string that contains
|
|
list of routes. A route is the mapping of domain and path to the exchange and queue name.
|
|
This method declare (and thus create) the queue and exchange, bind them together using the
|
|
regex-like routing expression given in the route.
|
|
After this the Service is listening on (consuming from) the queue for incoming messages.
|
|
"""
|
|
if self.router.is_open(self.channel):
|
|
queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
|
|
logging.info(" [*] RabbitMQClient register routes, cfg mapping: [%s]", route_mapping)
|
|
# convert the comma-separated string into list of AMQRoute objects
|
|
requested_routes = self.router.unwrap_route_list(route_mapping)
|
|
for route in requested_routes:
|
|
try:
|
|
# Note: AMQRoute allows one of the exchange or queue to be None, in which case the default is used
|
|
queue_name = route.queue if route.queue else self.router.get_consume_queue_name()
|
|
logging.info(" [*] RabbitMQClient register route: [%s]", route)
|
|
queue: AbstractRobustQueue = await self.channel.declare_queue(name=queue_name,
|
|
durable=True,
|
|
exclusive=False,
|
|
auto_delete=False,
|
|
arguments=queue_args)
|
|
_exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE
|
|
exchange: AbstractRobustExchange = await self.channel.declare_exchange(
|
|
name=_exchange_name, type=ExchangeType.topic
|
|
)
|
|
await queue.bind(exchange, route.key)
|
|
_c_tag = await queue.consume(callback, no_ack=False)
|
|
# Register the route with the router and publish its detail to other adapters
|
|
await self.router.register_route(
|
|
AMQRoute(
|
|
route.component_name, _exchange_name, queue_name,
|
|
sanitize_as_rabbitmq_name(route.key),
|
|
route.timeout, route.valid_until
|
|
)
|
|
)
|
|
logging.info(" [%s] AMQ connection initialized for route: %s, listens on: %s/%s",
|
|
_c_tag, route, exchange, queue_name)
|
|
except Exception as err:
|
|
logging_error("Callback registration INCOMPLETE, %s", err)
|
|
else:
|
|
logging.warning(" [W] Channel is null, cannot register routes.")
|
|
|
|
async def register_data_reply_callback(self, rpc_callback):
|
|
_c_tag = await self.reply_to_queue.consume(callback=rpc_callback, no_ack=False)
|
|
logging.info(" [%s] RabbitMQClient register data reply callback on queue: %s", _c_tag,self.reply_to_queue.name)
|
|
|
|
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_internal_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: DataMessage):
|
|
"""
|
|
The API routine that sends a datagram message (DataMessage) to the destination. The Destination
|
|
is determined from the DataMessage 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=DataMessageFactory.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=DataMessageFactory.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)
|