140 lines
4.7 KiB
Python
140 lines
4.7 KiB
Python
from os import getenv
|
|
from logging import Logger
|
|
from punq import Container
|
|
from pyee import EventEmitter
|
|
from pyee.asyncio import AsyncIOEventEmitter
|
|
from aioredlock import Aioredlock
|
|
from redis.asyncio import Redis as AsyncRedis
|
|
|
|
from kernel.application.manager.stop_event import StopEvent as StopEventManager
|
|
from kernel.application.stop_event import StopEvent
|
|
from kernel.application.view.application import Application
|
|
from kernel.application.view.configuration import Configuration
|
|
from kernel.application.view.rabbitmq import RabbitMQ
|
|
from kernel.application.view.redis import Redis as RedisConfig
|
|
|
|
from kernel.application.kernel import Kernel
|
|
from shared.application.service.environment_loader import EnvironmentLoader
|
|
from shared.application.manager.listener_arguments_parser import ListenerArgumentsParser
|
|
from shared.application.factory.logger import Logger as LoggerFactory
|
|
from shared.application.view.listener_arguments import ListenerArguments
|
|
from kernel.rabbitmq.client import Client
|
|
from kernel.rabbitmq.view.authentication_params import AuthenticationParams
|
|
from kernel.rabbitmq.view.connection_params import ConnectionParams
|
|
|
|
|
|
class RabbitMQKernel(Kernel):
|
|
__container: Container
|
|
|
|
def __init__(self) -> None:
|
|
self.__load_env()
|
|
self.__init_container()
|
|
self.__register_config()
|
|
self.__register_arguments()
|
|
self.__register_stop_event()
|
|
self.__register_logger()
|
|
self.__register_rabbitmq_client()
|
|
self.__register_redis()
|
|
self.__register_lock_provider()
|
|
self.__register_event_emitter()
|
|
|
|
def get_container(self) -> Container:
|
|
return self.__container
|
|
|
|
def __load_env(self) -> None:
|
|
env_loader = EnvironmentLoader(override=False)
|
|
env_loader.load()
|
|
|
|
def __init_container(self) -> None:
|
|
self.__container = Container()
|
|
|
|
def __register_config(self) -> None:
|
|
application = Application(
|
|
name=getenv("APP_NAME", ""),
|
|
env=getenv("APP_ENV", ""),
|
|
)
|
|
|
|
rabbitmq = RabbitMQ(
|
|
host=getenv("RABBITMQ_HOST", "localhost"),
|
|
port=int(getenv("RABBITMQ_PORT", "5672")),
|
|
user=getenv("RABBITMQ_USER", "guest"),
|
|
password=getenv("RABBITMQ_PASS", "guest"),
|
|
vhost=getenv("RABBITMQ_VHOST", "/")
|
|
)
|
|
|
|
redis = RedisConfig(
|
|
host=getenv("REDIS_HOST", "localhost"),
|
|
port=int(getenv("REDIS_PORT", "6379")),
|
|
db=int(getenv("REDIS_DB", "0"))
|
|
)
|
|
|
|
config = Configuration(application, rabbitmq, redis)
|
|
|
|
self.__container.register(Configuration, instance=config)
|
|
|
|
def __register_arguments(self) -> None:
|
|
argument_parser = ListenerArgumentsParser()
|
|
arguments_parsed = argument_parser.parse_args()
|
|
|
|
arguments = ListenerArguments(
|
|
queue=arguments_parsed.queue,
|
|
exchange=arguments_parsed.exchange,
|
|
routing_key=arguments_parsed.routing_key
|
|
)
|
|
|
|
self.__container.register(ListenerArguments, instance=arguments)
|
|
|
|
def __register_stop_event(self) -> None:
|
|
stop_event = StopEvent()
|
|
|
|
stop_event_manager = StopEventManager(stop_event)
|
|
stop_event_manager.setup_stop_signals()
|
|
|
|
self.__container.register(StopEvent, instance=stop_event)
|
|
|
|
def __register_logger(self) -> None:
|
|
logger_factory = LoggerFactory()
|
|
logger = logger_factory.create_stdout(name="rbmq.listener")
|
|
|
|
self.__container.register(Logger, instance=logger)
|
|
|
|
def __register_rabbitmq_client(self) -> None:
|
|
config = self.__container.resolve(Configuration).rabbitmq()
|
|
|
|
rabbitmq_client = Client(
|
|
AuthenticationParams(config.user, config.password),
|
|
ConnectionParams(config.host, config.port, config.vhost),
|
|
self.__container.resolve(Logger)
|
|
)
|
|
|
|
self.__container.register(Client, instance=rabbitmq_client)
|
|
|
|
def __register_redis(self) -> None:
|
|
config = self.__container.resolve(Configuration).redis()
|
|
|
|
client = AsyncRedis(
|
|
host=config.host,
|
|
port=config.port,
|
|
db=config.db,
|
|
decode_responses=True
|
|
)
|
|
|
|
self.__container.register(AsyncRedis, instance=client)
|
|
|
|
def __register_lock_provider(self) -> None:
|
|
config = self.__container.resolve(Configuration).redis()
|
|
|
|
aioredlock = Aioredlock([
|
|
{
|
|
"host": config.host,
|
|
"port": config.port,
|
|
"db": config.db
|
|
}
|
|
])
|
|
|
|
self.__container.register(Aioredlock, instance=aioredlock)
|
|
|
|
def __register_event_emitter(self) -> None:
|
|
event_emitter = AsyncIOEventEmitter()
|
|
self.__container.register(EventEmitter, instance=event_emitter)
|