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

import logging
from typing import Set

from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractIncomingMessage
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import CleverMicroServiceMessage, AMQRoute
from amqp.model.service_message_type import ServiceMessageType
from amqp.router.router_base import CM_C_REQ_QUEUE, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, ANY_RECIPIENT
from amqp.router.router_producer import RouterProducer


class RouterConsumer(RouterProducer):
    """
     * Contains Consumer logic. The logic here:
     * Sets a listener on CM_C_REQ_QUEUE queue to receive routing data requests
     * Upon receipt of routing data request, responds on CM_RESPONSE_EXCHANGE with local routing data.
     * It also proactively pushes the local routing data to CM_RESPONSE_EXCHANGE on startup to
     * notify the other (excluding self) Producers about its presence.
     *
     * Consumer is also Producer (Consumer sits alongside the CM Service, and anytime the Service needs
     * to communicate with other Service, it can do so via the Adapter). Therefore, the Producer
     * capability is also required.
    """

    def __init__(self, configuration: AMQConfiguration):
        super().__init__(configuration)
        self.amqRouteFactory = AMQRouteFactory()
        logging.info(" [*] RouterConsumer.<init>")

    async def init_routing_paths_consumer(self, _channel: AbstractRobustChannel):
        """
         * Initialize router according to Consumer mode.
        """
        logging.info(" [*] RouterConsumer.initConsumerRouter, channel open: %s", self.is_open(_channel))
        if self.is_open(_channel):
            try:
                #  1. declare a Queue to receive the RoutingData requests
                cm_request_queue: str = self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_REQ_QUEUE
                queue: AbstractRobustQueue = await self.channel.declare_queue(name=cm_request_queue, durable=True, exclusive=False, auto_delete=False)
                # 1a. bind queue to CM_REQUEST_EXCHANGE w/o any conditions (empty routing key)
                await queue.bind(exchange=CM_REQUEST_EXCHANGE, routing_key="")
                # 1b. Service adapter mode - listens on Request queue and replies to Response queue
                logging.info(" [*] RouterConsumer listens for RouteDataRequests on: %s", cm_request_queue)
                await queue.consume(callback=self.handle_routing_data_request_message, no_ack=False)
            except Exception as ioe:
                logging.error(" [E] RouterConsumer FATAL: cannot initialize, %s", ioe)
        else:
            logging.error(" [E] RouterConsumer FATAL: cannot initialize, RabbitMQ channel is not open.")

    """
     * ServiceMessageType.ROUTING_DATA_REQUEST  handler  listening on queue 'CleverMicroRequest'
    """

    async def handle_routing_data_request_message(self, message: AbstractIncomingMessage):
        # some debug statements, TODO - remove for prod release
        delivery_tag = message.delivery_tag
        debug = "Delivery.Properties = " + message.properties.__str__()
        logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s",
                     delivery_tag, message.consumer_tag, message.properties, debug)

        await message.ack(False)

        cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
        route_mapping: str = self.wrap_own_routes()
        logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]",
                     delivery_tag, self.service_message_factory.to_string(cm_message), route_mapping)
        # Reply only if self adapter has own routeMapping (e.g. API-gateway-side adapter has none)
        if cm_message.is_valid() and cm_message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping:
            if cm_message.reply_to != self.amq_configuration.amq_adapter.service_name:
                try:
                    service_message: CleverMicroServiceMessage = self.service_message_factory.of(
                        ServiceMessageType.ROUTING_DATA, route_mapping, cm_message.reply_to
                    )
                    await self.publish_service_message(service_message, self.cm_response_exchange)
                except Exception as err:
                    logging.error(" [E] ******[%s] (ROUTING_DATA_REQ) Response to queue: %s failed: %s",
                                  delivery_tag, CM_RESPONSE_EXCHANGE, err)
            else:
                logging.info(" [*] ******[%s] (ROUTING_DATA_REQ) [%s] Ignore request originated by ourselves.")
        else:
            logging.error(" [E] ******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.",
                          delivery_tag, str(message.body))

    async def register_route(self, route: AMQRoute):
        """
         * Called in Consumer (Service Adapter) context. Registers the routing data for self Service
         * with the Master router by sending own routing data to Master via CleverMicroResponse queue.
         * :param route: route to register with Master
        """
        try:
            if self.is_open(self.channel) and route.queue:
                serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
                    ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT
                )
                await self.publish_service_message(serviceMessage, self.cm_response_exchange)
                self.add_own_route(route)
                logging.info(" [DBG] RouterConsumer registered Route: %s", route)
            else:
                logging.warning(" [W] RouterConsumer couldn't register route data: %s channel isn't open.",
                                CM_RESPONSE_EXCHANGE)
        except Exception as ioe:
            logging.error(" [E] RouterConsumer failed register route data: %s", ioe)

    async def remove_route_on_cleanup(self, route: AMQRoute) -> bool:
        """
         * Unregister / remove route from the list
         * :param route: route to remove
         * :return result of remove operation
        """
        logging.info(" [DBG] RouterConsumer.removeRoute, route: %s", route.as_string())
        try:
            service_message: CleverMicroServiceMessage = self.service_message_factory.of(
                ServiceMessageType.ROUTING_DATA_REMOVE,
                route.as_string(),
                ANY_RECIPIENT
            )
            if not await self.publish_service_message(service_message, self.cm_response_exchange):
                logging.warning(" [W] RouterConsumer couldn't remove current route: %s, channel not open.",
                                route)

        except Exception as ioe:
            logging.error(" [E] RouterConsumer failed remove current route: %s", ioe)

        result: bool = self.remove_route(route)
        return result

    def cleanup(self):
        """
         * Cleanup - de-register routes with master.
        """
        routes: Set[AMQRoute] = self.get_routes()
        logging.info(" [DBG] RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes())
        not_removed = sum(1 for r in (self.remove_route_on_cleanup(route) for route in routes) if not r)
        if not_removed > 0:
            logging.info(" [W] %d route(s) not removed. self may lead to failed delivery on self route", not_removed)