Source code for amqp.router.route_database

import re
from typing import Optional, Set

from amqp.adapter.logging_utils import logging_info
from amqp.model.model import AMQRoute
from amqp.router.utils import sanitize_as_rabbitmq_name




class RouteDatabase:


    def __init__(self):
        self.routes: Set[AMQRoute] = set()




    def get_routes(self) -> Set[AMQRoute]:
        return self.routes




    def pattern(self, routing_key: str) -> re.Pattern:
        breaker = False
        tokens = routing_key.split(".")
        counter = len(tokens)
        regex_parts = []

        for token in tokens:
            counter -= 1
            if not breaker:
                if token == "":
                    res = "[^\\.]*"
                elif token == "#":
                    breaker = True
                    res = ".*$"
                else:
                    res = token
                if counter > 0 and not breaker:
                    res += "\\."
                regex_parts.append(res)
            else:
                regex_parts.append("")

        regex = "".join(filter(lambda x: x != "", regex_parts))
        if not routing_key:
            return re.compile(".*")
        return re.compile(regex)




    def matches(self, routing_key_from_route: str, routing_key_from_message: str) -> bool:
        route_pattern = self.pattern(routing_key_from_route)
        return bool(route_pattern.match(sanitize_as_rabbitmq_name(routing_key_from_message)))




    def wrap_routes(self) -> str:
        return ",".join(str(route) for route in self.routes)




    def find_route(self, domain: str, path: str) -> AMQRoute | None:
        routing_key = sanitize_as_rabbitmq_name(f"{domain}.{path}")
        logging_info(f"findRoute key:{routing_key} in:{self.wrap_routes()}")
        for route in self.routes:
            if self.matches(route.key, routing_key):
                return route
        return None




    def find_route_by_service(self, service_name: str) -> Optional[AMQRoute]:
        logging_info(f"findRoute service_name:{service_name} in:{self.wrap_routes()}")
        for route in self.routes:
            if route.component_name == service_name:
                return route
        return None




    def add_route(self, route: AMQRoute) -> bool:
        if route not in self.routes:
            self.routes.add(route)
            return True
        return False




    def remove_route(self, route: AMQRoute) -> bool:
        return route in self.routes and self.routes.remove(route)