from typing import Callable, List, Optional, Set
from amqp.adapter.amq_route_factory import NULL_ROUTE, AMQRouteFactory
from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
from amqp.model.model import AMQRoute, DataMessage
from amqp.router.route_database import RouteDatabase
from amqp.router.utils import sanitize_as_rabbitmq_name
CM_REQUEST_EXCHANGE = "CM-Request"
CM_RESPONSE_EXCHANGE = "CM-Response"
CM_C_REQ_QUEUE = "C-REQ"
CM_C_RESP_QUEUE = "C-RESP"
CM_C_AMQ_QUEUE = "C-AMQ"
CM_P_RESP_QUEUE = "P-RESP"
CM_P_REPLY_TO = "P-REPLY-TO"
ANY_RECIPIENT = "*"
AMQ_RPC_EXCHANGE = "AMQ-RPC"
AMQ_REPLY_TO_EXCHANGE = "AMQ-REPLY-TO"
DLQ_EXCHANGE = "DLX"
[docs]
class RouterBase:
[docs]
def __init__(self):
self.route_database = RouteDatabase()
self.own_routes: Set[AMQRoute] = set()
self.route_added_notifier: Callable[[AMQRoute], None] | None = None
self.route_removed_notifier: Callable[[AMQRoute], None] | None = None
[docs]
def get_routing_key(self, message: DataMessage) -> str:
return sanitize_as_rabbitmq_name(f"{message.domain()}.{message.path()}")
[docs]
def wrap_routes(self) -> str:
return self.route_database.wrap_routes()
[docs]
def wrap_own_routes(self) -> str:
return ",".join(str(route) for route in self.own_routes)
[docs]
def unwrap_route_list(self, route_string: str) -> List[AMQRoute]:
logging_debug("RouterMaster.unwrapRoutes, route(s): %s", route_string)
routes = [AMQRouteFactory.from_string(route_str) for route_str in route_string.split(",")]
return [route for route in routes if route != NULL_ROUTE]
[docs]
def get_routes(self) -> Set[AMQRoute]:
return self.route_database.get_routes()
[docs]
def find_route(self, domain: str, path: str) -> AMQRoute | None:
return self.route_database.find_route(domain, path)
[docs]
def find_route_by_message(self, message: DataMessage) -> AMQRoute | None:
return self.route_database.find_route(message.domain(), message.path())
[docs]
def find_route_by_service(self, service_name: str) -> Optional[AMQRoute]:
_route: Optional[AMQRoute] = self.route_database.find_route_by_service(service_name)
if _route is None:
for route in self.own_routes:
if route.component_name == service_name:
_route = route
return _route
[docs]
def add_routes(self, message: str) -> None:
try:
amq_route_list = self.unwrap_route_list(message)
logging_info(" [DBG] RouterMaster.addRoute [%d]: %s", len(amq_route_list), message)
for route in amq_route_list:
logging_info("RouterMaster.addRoute, adding route: %s", route)
if not route.exchange and not route.queue and not route.key:
logging_error(
"Cannot setup AMQ Exchange for '%s', at least one of "
"Exchange, Queue and/or Key must be present in the routing string. "
"Expected format is '%s'",
route,
AMQRoute.FORMAT,
)
else:
if route.exchange != DLQ_EXCHANGE:
is_new_route = self.route_database.add_route(route)
if is_new_route and self.route_added_notifier:
self.route_added_notifier(route)
except Exception as err:
logging_error("add_routes error: %s", err)
[docs]
def set_route_added_notifier(self, callback: Callable[[AMQRoute], None]):
self.route_added_notifier = callback
[docs]
def remove_routes(self, message: str) -> None:
try:
to_remove = [AMQRouteFactory.from_string(route_str) for route_str in message.split(",")]
removed_count = sum(1 for route in to_remove if self.remove_route(route))
logging_info(f"RouterMaster.removeRoutes(): {message}, removed={removed_count}")
except IOError as err:
logging_error(f"Can't remove route(s): {err}")
[docs]
def remove_route(self, route: AMQRoute) -> bool:
is_route_removed = self.route_database.remove_route(route)
if is_route_removed and self.route_removed_notifier:
self.route_removed_notifier(route)
return is_route_removed
[docs]
def add_route(self, route: AMQRoute) -> None:
self.route_database.add_route(route)
[docs]
def add_own_route(self, route: AMQRoute) -> None:
self.own_routes.add(route)