Stanislav Hejny
85 lines
3.8 KiB
Python
85 lines
3.8 KiB
Python
import os
|
|
import logging
|
|
from functools import partial
|
|
from opentelemetry import trace
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import (
|
|
ConsoleSpanExporter,
|
|
SimpleSpanProcessor
|
|
)
|
|
from opentelemetry.propagate import set_global_textmap
|
|
from opentelemetry.propagators.tracecontext import TraceContextTextMapPropagator
|
|
|
|
from amqp.adapter.amq_message_factory import AMQMessageFactory
|
|
from amqp.adapter.amq_route_factory import AMQRouteFactory
|
|
from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
|
|
from amqp.rabbitmq.rabbit_mq_consumer import RabbitMQConsumer
|
|
from amqp.service.service_message_handler import AMQServiceMessageHandler
|
|
|
|
|
|
class AMQService:
|
|
CLEVER_AMQP_CLIENT_VERSION = os.getenv("CLEVER_AMQP_CLIENT_VERSION")
|
|
MAX_WAIT = 300 # defaults to 5 minutes (300 seconds)
|
|
|
|
def __init__(self, amq_configuration, http_adapter):
|
|
self.amq_configuration = amq_configuration
|
|
self.http_adapter = http_adapter
|
|
self.message_factory = AMQMessageFactory.get_instance(self.amq_configuration.amq_adapter().generator_id())
|
|
self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration)
|
|
self.amq_service_message_handler = AMQServiceMessageHandler(self.http_adapter, self.get_tracer())
|
|
|
|
def init(self):
|
|
CleverMicroServiceMessageFactory.process_name(self.amq_configuration.amq_adapter().service_name())
|
|
logging.info("********* AMQService.isRouter = %s ***********", self.amq_configuration.amq_adapter().is_router())
|
|
logging.info("***********************************************")
|
|
logging.info("********** AMQ Service *******************")
|
|
logging.info("***********************************************")
|
|
|
|
self.setup_opentelemetry()
|
|
self.register_routes()
|
|
self.register_rpc_handler()
|
|
self.rabbit_mq_consumer.request_routes()
|
|
|
|
def cleanup(self):
|
|
logging.info(" [DBG] AMQService.cleanup() rmqc=%s", self.rabbit_mq_consumer)
|
|
if self.rabbit_mq_consumer:
|
|
self.rabbit_mq_consumer.cleanup()
|
|
|
|
def reinitialize_if_disconnected(self):
|
|
if not self.rabbit_mq_consumer.channel or not self.rabbit_mq_consumer.channel.is_open:
|
|
self.init()
|
|
|
|
def setup_opentelemetry(self):
|
|
trace.set_tracer_provider(TracerProvider())
|
|
trace.get_tracer_provider().add_span_processor(
|
|
SimpleSpanProcessor(ConsoleSpanExporter())
|
|
)
|
|
set_global_textmap(TraceContextTextMapPropagator())
|
|
self.tracer = trace.get_tracer("com.cleverthis")
|
|
logging.info("OTEL %s %s %s", trace.get_logger_bridge(), trace.get_meter_provider(), trace.get_tracer_provider())
|
|
logging.info("OTEL TRACER %s", trace.get_propagators())
|
|
|
|
def register_routes(self):
|
|
route_mapping = self.amq_configuration.amq_adapter().route_mapping()
|
|
if AMQRouteFactory.validate(route_mapping):
|
|
try:
|
|
self.rabbit_mq_consumer.register_routes()
|
|
self.rabbit_mq_consumer.register_deliver_callback(
|
|
partial(self.amq_service_message_handler.on_delivery_callback_http, self.rabbit_mq_consumer.channel)
|
|
)
|
|
self.rabbit_mq_consumer.register_rpc_handler(
|
|
self.amq_service_message_handler.on_reply_consumer(self.rabbit_mq_consumer.channel)
|
|
)
|
|
except Exception as e:
|
|
raise RuntimeError(e)
|
|
elif route_mapping:
|
|
logging.info(" [W] route [%s] failed validation, no routes were registered.", route_mapping)
|
|
|
|
def register_rpc_handler(self):
|
|
try:
|
|
self.rabbit_mq_consumer.register_rpc_handler(
|
|
self.amq_service_message_handler.on_reply_consumer(self.rabbit_mq_consumer.get_channel())
|
|
)
|
|
except Exception as err:
|
|
raise
|