"""
* Maps the request to RabbitMQ exchange. Handles the service discovery logic.
*
* Router can be either in 'Master' or 'Service (Consumer)' modes. 'Master' mode is the outward
* facing side that receives the requests and routes it to queue associated with requested service.
* 'Service' mode is consumer side for given service, which receives the message from AMQ and
* forwards it to the application.
"""
from aio_pika import Message
from aio_pika.abc import (
AbstractIncomingMessage,
AbstractRobustChannel,
AbstractRobustConnection,
AbstractRobustExchange,
AbstractRobustQueue,
)
from pika.exchange_type import ExchangeType
from amqp.adapter.logging_utils import (
logging_debug,
logging_error,
logging_info,
logging_warning,
)
from amqp.adapter.service_message_factory import ServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import ServiceMessage
from amqp.model.service_message_type import ServiceMessageType
from amqp.router.router_base import (
ANY_RECIPIENT,
CM_C_AMQ_QUEUE,
CM_P_REPLY_TO,
CM_P_RESP_QUEUE,
CM_REQUEST_EXCHANGE,
CM_RESPONSE_EXCHANGE,
RouterBase,
)
[docs]
class RouterProducer(RouterBase):
[docs]
def __init__(self, configuration: AMQConfiguration):
"""
* Initialize router - need to supply the config values and RabbitMQ channel.
* :param configuration: adapter's configuration reference.
"""
super().__init__()
self.connection: AbstractRobustConnection | None = None
self.service_message_factory = ServiceMessageFactory(configuration.amq_adapter.service_name)
self.amq_configuration: AMQConfiguration = configuration
self.channel: AbstractRobustChannel | None = None
self.cm_request_exchange: AbstractRobustExchange | None = None
self.cm_response_exchange: AbstractRobustExchange | None = None
logging_info(
"RouterProducer.<init>: %s, route: %s",
self.amq_configuration.amq_adapter.service_name,
self.amq_configuration.amq_adapter.route_mapping,
)
[docs]
async def init_internal_routing_paths(self, channel: AbstractRobustChannel | None):
"""
* Initialize internal routes for ServiceMessages.
* The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues
* created to enable the service discovery.
* The Producer mode sends out routing data request on CleverMicroRequest and listens
* on CleverMicroResponse for any service responding with its routing data.
"""
self.channel = channel
logging_info(
"RouterProducer.initRoutingPaths, channel open: %s",
self.is_open(self.channel),
)
if self.is_open(self.channel):
try:
# **** set up the service discovery internal exchanges ****
# 1. declare RequestExchange to where to publish the RoutingData requests
self.cm_request_exchange = await self.channel.declare_exchange(
name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout
)
#
# 2. declare ResponseExchange where to publish current_availability RoutingData (response to
# a RoutingData request)
self.cm_response_exchange = await self.channel.declare_exchange(
name=CM_RESPONSE_EXCHANGE, type=ExchangeType.fanout
)
# 2a. declare a Response queue for Routing Data replies from Services
_cm_response_queue: str = self.get_response_queue_name()
_response_queue: AbstractRobustQueue = await self.channel.declare_queue(
name=_cm_response_queue,
durable=True,
exclusive=False,
auto_delete=False,
arguments={},
)
# 2b. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key)
await _response_queue.bind(exchange=CM_RESPONSE_EXCHANGE, routing_key="#")
# 2c. listen for incoming RoutingData messages
_c_tag = await _response_queue.consume(
callback=self.handle_routing_data_message, no_ack=False
)
logging_info(
" [%s] Routing master listens for Route updates on %s",
_c_tag,
_cm_response_queue,
)
except Exception as ioe:
logging_error("RouterProducer FATAL: cannot initialize, %s", ioe)
else:
logging_error("RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.")
[docs]
async def handle_routing_data_message(self, message: AbstractIncomingMessage):
"""
* Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of
* routing data, invokes addRoute or removeRoute depending on the type of the message.
"""
logging_info(
"[%s] Received ROUTING DATA: [%s], msgId=%s",
message.delivery_tag,
message.body,
message.message_id,
)
await message.ack(multiple=False)
cm_message: ServiceMessage = self.service_message_factory.from_bytes(message.body)
# ignore the message if it comes from ourselves (even if from different instance)
logging_debug(
"******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s",
message.delivery_tag,
self.service_message_factory.to_string(cm_message),
self.amq_configuration.amq_adapter.route_mapping,
self.amq_configuration.amq_adapter.service_name,
cm_message.reply_to,
)
if (
cm_message.is_valid()
and not self.amq_configuration.amq_adapter.service_name == cm_message.reply_to
):
if cm_message.message_type == ServiceMessageType.ROUTING_DATA:
self.add_routes(cm_message.message)
if cm_message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE:
self.remove_routes(cm_message.message)
[docs]
async def request_routes_from_adapters(self):
"""
* Send a broadcast message prompting all listening adapters to reply with own routing data.
"""
logging_info(
"RouterProducer.requestRoutesFromAdapters, queue name: %s",
CM_REQUEST_EXCHANGE,
)
try:
serviceMessage: ServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
)
if not await self.publish_service_message(serviceMessage, self.cm_request_exchange):
logging_warning(
"RouterProducer could not request route data: %s channel is not open.",
CM_REQUEST_EXCHANGE,
)
except Exception as ioe:
logging_error("RouterProducer failed request routing data: %s", ioe)
[docs]
async def publish_service_message(
self, message: ServiceMessage, exchange: AbstractRobustExchange
) -> bool:
"""
* Publishes a service message to the service queue and logs success log entry.
*
* :param message: Service Message to be sent
* :param exchange: target exchange name where to publish the message
* :return True if channel is open, False otherwise
* @throws IOException if something else goes wrong
"""
if self.is_open(self.channel):
binary_content: bytes = self.service_message_factory.serialize(message)
pika_message: Message = Message(
body=binary_content,
content_encoding="utf-8",
delivery_mode=2,
content_type="application/octet-stream",
headers=None,
priority=0,
correlation_id=None,
)
await exchange.publish(message=pika_message, routing_key="")
logging_info(
"Service Message Published to %s, msg: %s",
exchange.name,
str(binary_content),
)
return True
return False
[docs]
def get_reply_to_queue_name(self) -> str:
"""
* self should create a unique name for self process and also when there's several instances.
* @return unique reply-to queue name.
"""
return self.amq_configuration.amq_adapter.adapter_prefix() + CM_P_REPLY_TO
[docs]
def get_response_queue_name(self) -> str:
"""
* self should create a unique name for self process and also when there's several instances.
* @return unique reply-to queue name.
"""
return self.amq_configuration.amq_adapter.adapter_prefix() + CM_P_RESP_QUEUE
[docs]
def get_consume_queue_name(self) -> str:
"""
* self should create a unique name for self process and also when there's several instances.
* @return unique reply-to queue name.
"""
return self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_AMQ_QUEUE
[docs]
def is_open(self, _channel: AbstractRobustChannel) -> bool:
"""
* Test if the channel is usable (opened).
* :param _channel: the channel
* :return True if the channel is opened.
"""
return _channel is not None and not _channel.is_closed