amq-adapter-python/amqp/router/router_producer.py

"""
 * 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.
 """
import logging

from aio_pika import Message
from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractRobustConnection, AbstractRobustExchange, \
    AbstractIncomingMessage
from pika.exchange_type import ExchangeType

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 RouterBase, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, CM_C_AMQ_QUEUE, \
    CM_P_RESP_QUEUE, CM_P_REPLY_TO, ANY_RECIPIENT


class RouterProducer(RouterBase):

    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(f" [*] RouterProducer.<init>: %s, route: %s",
                     self.amq_configuration.amq_adapter.service_name,
                     self.amq_configuration.amq_adapter.route_mapping)

    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 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(" [E] RouterProducer FATAL: cannot initialize, %s", ioe)

        else:
            logging.error(" [E] RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.")

    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(f" [x] Received {message.body}, m={message.message_id}, p={message.properties}")
        """
        logging.info(" [DBG] ******[%s] ROUTING_DATA %s >>> %s",
              delivery.getEnvelope().getDeliveryTag(),
              delivery.getEnvelope().toString(), str(delivery.getBody()))
              """
        logging.info(" [%s] ACK message now so that it is not being redelivered", message.delivery_tag)
        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.info(" [DBG] ******[%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)

    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(" [E] RouterProducer could not request route data: %s channel is not open.",
                                CM_REQUEST_EXCHANGE)

        except Exception as ioe:
            logging.error(" [E] RouterProducer failed request routing data: %s", ioe)

    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(" [x] Service Message Sent to %s, msg: %s", exchange.name, str(binary_content))
            return True
        return False

    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

    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

    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

    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