From c4037f3cd905726d96803891b74d72329a9e76c5 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Sun, 2 Mar 2025 00:35:09 +0000 Subject: [PATCH 1/7] Fix issues from testing --- amqp/adapter/cleverthis_service_adapter.py | 15 ++++++++++++++- amqp/service/service_message_handler.py | 5 +++-- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index d34d68f..d00af1e 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -6,6 +6,7 @@ import re from aiohttp import ClientSession, ClientResponse from fastapi import HTTPException, UploadFile +from python_multipart import multipart from python_multipart.multipart import Field, File from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any from urllib.parse import parse_qs @@ -21,7 +22,19 @@ class CleverMultiPartParser: self.form_data = {} # Convert each list of strings to a single string joined by newline, then to bytes headers = {key: "\n".join(value).encode("utf-8") for key, value in message.headers.items()} - python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file) + content_type: str | bytes | None = headers.get("Content-Type") + content_type, params = multipart.parse_options_header(content_type) + if content_type in ["application/octet-stream", + "multipart/form-data", + "application/x-www-form-urlencoded", + "application/x-url-encoded"]: + try: + python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file) + except Exception as e: + logging.error(f"Error parsing multipart/form-data: {e}") + self.form_data = message.body() + else: + self.form_data = message.body() def on_field(self, field: Field): print(field) diff --git a/amqp/service/service_message_handler.py b/amqp/service/service_message_handler.py index b977be7..23a4328 100644 --- a/amqp/service/service_message_handler.py +++ b/amqp/service/service_message_handler.py @@ -67,7 +67,7 @@ class AMQServiceMessageHandler: amq_message: AMQMessage = AMQMessageFactory.from_bytes(body) logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s", - delivery_tag, method.consumer_tag, amq_message.id, map_as_string(amq_message.trace_info)) + delivery_tag, method.consumer_tag, amq_message.id, map_as_string(amq_message.trace_info)) propagator = TraceContextTextMapPropagator() context = propagator.extract(carrier=amq_message.trace_info, context=trace.context_api.get_current()) @@ -135,10 +135,11 @@ class AMQServiceMessageHandler: """ if reply_to: logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s", - delivery_tag, reply_to, response.id) + delivery_tag, reply_to, response.id) reply_props = pika.BasicProperties( correlation_id=response.id, content_type=response.content_type[0] + if isinstance(response.content_type, list) else response.content_type ) channel.basic_publish( exchange=AMQ_REPLY_TO_EXCHANGE, -- 2.52.0 From 1ffd475e52afe12d080504fcf695186b0e71cae0 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Sun, 2 Mar 2025 00:43:22 +0000 Subject: [PATCH 2/7] Fix issues from testing --- pyproject.toml | 2 +- setup.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 750bec3..b998c7a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.2" +version = "0.2.3" description = "The Python implementation of the AMQ Adaptor for CleverMicro" readme = "README.md" license = { text = "Apache License 2.0" } diff --git a/setup.py b/setup.py index db0d66d..a976db4 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import setup, find_packages setup( name="amqp", - version="0.2.0", + version="0.2.3", author="Stanislav Hejny", author_email="stanislav.hejny@cleverthis.com", description="Adapter for RabbitMQ to integrate a CleverThis python-code Service with CleverMicro platform", -- 2.52.0 From 2e1b95675a29d00f75f1a8f952f4e643b6975258 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Sun, 2 Mar 2025 12:37:20 +0000 Subject: [PATCH 3/7] fix the byte to string conversion --- amqp/adapter/cleverthis_service_adapter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index d00af1e..75f7b8c 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -24,7 +24,7 @@ class CleverMultiPartParser: headers = {key: "\n".join(value).encode("utf-8") for key, value in message.headers.items()} content_type: str | bytes | None = headers.get("Content-Type") content_type, params = multipart.parse_options_header(content_type) - if content_type in ["application/octet-stream", + if content_type.decode('utf-8') in ["application/octet-stream", "multipart/form-data", "application/x-www-form-urlencoded", "application/x-url-encoded"]: -- 2.52.0 From a7d7832a4904f6a15bb034685c4c3f870e0d288e Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Sun, 2 Mar 2025 12:50:46 +0000 Subject: [PATCH 4/7] fix the byte to string conversion --- pyproject.toml | 2 +- setup.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index b998c7a..124e2bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.3" +version = "0.2.4" description = "The Python implementation of the AMQ Adaptor for CleverMicro" readme = "README.md" license = { text = "Apache License 2.0" } diff --git a/setup.py b/setup.py index a976db4..df6c0be 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import setup, find_packages setup( name="amqp", - version="0.2.3", + version="0.2.4", author="Stanislav Hejny", author_email="stanislav.hejny@cleverthis.com", description="Adapter for RabbitMQ to integrate a CleverThis python-code Service with CleverMicro platform", -- 2.52.0 From 5985a22bbb93c0e3d82b3ac52de8b00c067ee6c6 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Thu, 13 Mar 2025 02:13:56 +0000 Subject: [PATCH 5/7] rewrite using asynch PIKA library --- amqp/adapter/cleverthis_service_adapter.py | 40 ++++++- amqp/rabbitmq/rabbit_mq_consumer.py | 129 ++++++--------------- amqp/rabbitmq/rabbit_mq_producer.py | 128 +++++++------------- amqp/router/router_consumer.py | 50 ++++---- amqp/router/router_producer.py | 113 +++++++++--------- amqp/service/amq_service.py | 39 +++++-- amqp/service/service_message_handler.py | 101 ++++++++-------- amqp/service/tracing.py | 1 - pyproject.toml | 2 +- requirements.txt | 2 +- setup.py | 2 +- 11 files changed, 263 insertions(+), 344 deletions(-) diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 75f7b8c..693aab1 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -1,8 +1,11 @@ +import asyncio import json import logging import python_multipart import io +import os import re +import threading from aiohttp import ClientSession, ClientResponse from fastapi import HTTPException, UploadFile @@ -16,6 +19,11 @@ from amqp.adapter.amq_response_factory import AMQResponseFactory from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQMessage, AMQResponse +# Track the number of messages currently being processed +current_parallel_executions = 0 +lock = threading.Lock() +parallel_workers = os.getenv("PARALLEL_WORKERS", default=10) + class CleverMultiPartParser: def __init__(self, message: AMQMessage): @@ -183,6 +191,14 @@ class CleverThisServiceAdapter: """ return None + def sync_on_message(self, message: AMQMessage) -> AMQResponse: + """ + Synchronous version of on_message method. This is a wrapper around the async on_message method. + :param message: AMQMessage + :return: AMQResponse + """ + return asyncio.run(self.on_message(message)) + async def on_message(self, message) -> AMQResponse: """ Called when a message is received from the AMQP. @@ -190,6 +206,14 @@ class CleverThisServiceAdapter: :return: AMQP response class with operation status code and optional error details. """ DEBUG_NO_AUTH = 1 + global current_parallel_executions + + # Increment the counter for parallel executions + with lock: + current_parallel_executions += 1 + if current_parallel_executions > parallel_workers - 2: + logging.warn("Warning: CApacity close to depleted!") + _auth: str = message.headers.get('Authorization', '') _auth_user: Any | None = None if 'Bearer' in _auth: @@ -198,22 +222,26 @@ class CleverThisServiceAdapter: _auth_user = await self.get_current_user(token) except Exception as error: logging.error(f"Error getting user: {error}") + amq_response: AMQResponse | None = None if _auth_user or DEBUG_NO_AUTH == 1: method = message.method.lower() if method == 'get': - return await self.get(message, _auth_user) + amq_response = await self.get(message, _auth_user) elif method == 'put': - return await self.put(message, _auth_user) + amq_response = await self.put(message, _auth_user) elif method == 'post': - return await self.post(message, _auth_user) + amq_response = await self.post(message, _auth_user) elif method == 'head': - return await self.head(message, _auth_user) + amq_response = await self.head(message, _auth_user) elif method == 'delete': - return await self.delete(message, _auth_user) + amq_response = await self.delete(message, _auth_user) else: raise ValueError(f"Unexpected HTTP method: {message.method}") else: - return AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info) + amq_response = AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info) + with lock: + current_parallel_executions -= 1 + return amq_response # ================================================================================================== # ===================== C l e v e r T h i s A P I m e t h o d s ============================ diff --git a/amqp/rabbitmq/rabbit_mq_consumer.py b/amqp/rabbitmq/rabbit_mq_consumer.py index 7d8880c..8c7270c 100644 --- a/amqp/rabbitmq/rabbit_mq_consumer.py +++ b/amqp/rabbitmq/rabbit_mq_consumer.py @@ -1,7 +1,8 @@ import logging import pika -from pika.exchange_type import ExchangeType +from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange +from pika.exchange_type import ExchangeType from amqp.model.model import AMQRoute from amqp.rabbitmq.rabbit_mq_producer import RabbitMQProducer from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE, DLQ_EXCHANGE, AMQ_RPC_EXCHANGE @@ -15,6 +16,20 @@ class RabbitMQConsumer(RabbitMQProducer): def __init__(self, configuration): super().__init__(configuration, RouterConsumer(configuration)) + self.reply_to_exchange: AbstractRobustExchange | None = None # when sending response to RPC request + self.reply_to_queue: AbstractRobustQueue | None= None # when receiving response to RPC request + + async def init(self, loop) -> AbstractRobustExchange: + await self.setup_amq_channel(loop) + cm_reply_to_queue = self.router.get_reply_to_queue_name() + self.reply_to_queue = await self.channel.declare_queue( + name=cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False + ) + self.reply_to_exchange = await self.channel.declare_exchange( + AMQ_REPLY_TO_EXCHANGE, type=pika.exchange_type.ExchangeType.direct + ) + await self.reply_to_queue.bind(exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue) + return self.reply_to_exchange def cleanup(self): logging.info(" [*] RabbitMQConsumer.cleanup()") @@ -22,118 +37,44 @@ class RabbitMQConsumer(RabbitMQProducer): if self.router.is_open(self.channel): try: self.channel.close() - except (pika.connection.exceptions.AMQPError, pika.connection.exceptions.ChannelClosedByBroker) as e: + except Exception as e: logging.error(" [E] RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e) - def register_inbound_routes(self, callback): + async def register_inbound_routes(self, route_mapping: str, callback): if self.router.is_open(self.channel): queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None - route_mapping = self.amq_configuration.amq_adapter.route_mapping logging.info(" [*] RabbitMQConsumer register routes, cfg mapping: [%s]", route_mapping) requested_routes = self.router.unwrap_route_list(route_mapping) for route in requested_routes: try: queue_name = route.queue if route.queue else self.router.get_consume_queue_name() logging.info(" [*] RabbitMQConsumer register route: [%s]", route) - self.channel.queue_declare(queue_name, durable=True, exclusive=False, - auto_delete=False, arguments=queue_args) - exchange = route.exchange if route.exchange else AMQ_RPC_EXCHANGE - self.channel.exchange_declare(exchange, exchange_type=pika.exchange_type.ExchangeType.topic) - self.channel.queue_bind(queue_name, exchange, route.key) - self.channel.basic_consume( - queue=queue_name, - on_message_callback=callback, - auto_ack=False + queue: AbstractRobustQueue = await self.channel.declare_queue(name=queue_name, + durable=True, + exclusive=False, + auto_delete=False, + arguments=queue_args) + exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE + exchange: AbstractRobustExchange = await self.channel.declare_exchange( + name=exchange_name, type=ExchangeType.topic ) - self.router.register_route( + await queue.bind(exchange, route.key) + await queue.consume(callback, no_ack=False) + await self.router.register_route( AMQRoute( - route.component_name, exchange, queue_name, + route.component_name, exchange_name, queue_name, sanitize_as_rabbitmq_name(route.key), route.timeout, route.valid_until ) ) logging.info(" [*] AMQ connection initialized for route: %s, listens on: %s/%s", - route, exchange, queue_name) - except pika.connection.exceptions.AMQPError as err: + route, exchange, queue_name) + except Exception as err: logging.error(" [E] Callback registration INCOMPLETE, %s", err) else: logging.warning(" [W] Channel is null, cannot register routes.") - def register_rpc_handler(self, rpc_consumer): - cm_reply_to_queue = self.router.get_reply_to_queue_name() - self.channel.queue_declare(cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False) - self.channel.exchange_declare( - AMQ_REPLY_TO_EXCHANGE, exchange_type=pika.exchange_type.ExchangeType.direct + async def register_rpc_handler(self, rpc_consumer): + await self.reply_to_queue.consume(callback=rpc_consumer, + no_ack=False ) - self.channel.queue_bind(cm_reply_to_queue, AMQ_REPLY_TO_EXCHANGE, cm_reply_to_queue) - self.channel.basic_consume( - queue=cm_reply_to_queue, - on_message_callback=rpc_consumer, - auto_ack=False - ) - - """ - def limit_rate_inside_time_window(self, message: AMQMessage) -> int: - next_batch_size = BackpressureSubscriber.NO_OP - now = time.time_ns() // 1000000 - queue_size = self.getQueueSize(message) - - # 1. Backpressure? if unprocessed count exceeds threshold, wait for some time - if queue_size > THRESHOLD: - # TODO - signal queue backpressure explicitly to upstream/metrics - # wait: estimated-per-message-processing-time * SQRT(over-the-limit) - self.wait_for_milliseconds( - (WINDOW / self.request_for_window_duration) * (queue_size - THRESHOLD) ** 0.5) - queue_size = self.getQueueSize(message) # refresh queue size after wait - - # 2. batch complete? determine next batch size and wait till window end. reset window. - if self.fulfilled >= self.request_for_window_duration: - next_batch_size = self.calculateNextBatchSize(queue_size) - self.request_for_window_duration = next_batch_size - remaining_time = (self.window_started_at + WINDOW) - now - if remaining_time > 0: - self.wait_for_milliseconds(remaining_time) - queue_size = self.getQueueSize(message) # refresh queue size after wait - self.window_started_at = time.time_ns() // 1000000 - self.fulfilled = 0 - - # 3. Window elapsed? Reset window & recalculate batch size - if now >= self.window_started_at + WINDOW: - if next_batch_size == BackpressureSubscriber.NO_OP: - next_batch_size = self.calculateNextBatchSize(queue_size) - self.request_for_window_duration = next_batch_size - self.window_started_at = time.time_ns() // 1000000 - self.fulfilled = 0 - - # 4. send the message from the current batch - self.send_message(message) - self.fulfilled += 1 - - # 5. report and setup state for the next call - end = time.time_ns() // 1000000 - self.on_next_timestamp = end - return next_batch_size - - def calculateNextBatchSize(self, queue_size: int) -> int: - next_batch_size = self.request_for_window_duration - if queue_size <= THRESHOLD: - # processing proceeds well, can increase the rate - next_batch_size += self.subscriber.increaseRate(self.request_for_window_duration) - logging.debug(f"INCREASE rate to {next_batch_size}. qSize={queue_size}") - else: - decrease = self.subscriber.backoff(queue_size - THRESHOLD) - if self.request_for_window_duration > decrease: - next_batch_size -= decrease - logging.debug(f"DECREASE rate to {next_batch_size}. qSize={queue_size}") - return next_batch_size - """ - """ - def getQueueSize(self, message: AMQMessage) -> int: - try: - route_queue_name = self.router.findRoute(message) - queue_name = route_queue_name.queue if route_queue_name else self.TASK_DISPATCH_QUEUE_NAME - logging.info(f"Get Queue size for {queue_name}") - return max(len(self.awaitingACK), self.channel.message_count(queue_name)) - except Exception: - return len(self.awaitingACK) - """ diff --git a/amqp/rabbitmq/rabbit_mq_producer.py b/amqp/rabbitmq/rabbit_mq_producer.py index bc32dd0..d67db11 100644 --- a/amqp/rabbitmq/rabbit_mq_producer.py +++ b/amqp/rabbitmq/rabbit_mq_producer.py @@ -1,10 +1,5 @@ import logging -import time - -import pika -from pika.adapters.blocking_connection import BlockingChannel, BlockingConnection -from pika.channel import Channel -from pika.connection import Connection +import aio_pika from amqp.adapter.amq_message_factory import AMQMessageFactory from amqp.config.amq_configuration import AMQConfiguration @@ -14,38 +9,22 @@ from amqp.router.router_producer import RouterProducer class RabbitMQProducer: PREFETCH_COUNT = 1 - THRESHOLD = 1000 - WINDOW = 5000 # 5 seconds def __init__(self, amq_configuration: AMQConfiguration, router: RouterProducer = None): - self.channel: BlockingChannel | None = None - self.connection: BlockingConnection | None = None - self.request_for_window_duration = None + self.channel: aio_pika.RobustChannel | None = None + self.connection: aio_pika.RobustConnection | None = None + self.rpc_exchange: aio_pika.RobustExchange | None = None + self.reply_exchange: aio_pika.RobustExchange | None = None self.amq_configuration = amq_configuration logging.info("*******************************************") - logging.info("******** RabbitMQProducer.init() ********** %s - %s - %s", - self.amq_configuration.dispatch.amq_host, - self.amq_configuration.dispatch.use_confirms, - self.WINDOW) + logging.info("******** RabbitMQProducer.init(): %s", + self.amq_configuration.dispatch.amq_host) logging.info("*******************************************") self.router = router or RouterProducer(amq_configuration) - # this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); - self.setup_amq_channel() - self.router.init_routing_paths(self.channel) - - # self.backpressure_per_exchange = defaultdict(lambda: Gauge( - # f"amqp.adapter.exchange.{exchange}", registry=CollectorRegistry())) - - self.router.set_route_added_notifier( - lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s", - route.exchange)) - - self.request_for_window_duration = self.THRESHOLD // 2 - self.window_started_at = time.time() - self.on_next_timestamp = self.window_started_at - self.fulfilled = 0 + async def init(self, loop): + await self.setup_amq_channel(loop) def cleanup(self): if self.connection: @@ -54,30 +33,27 @@ class RabbitMQProducer: except Exception as e: logging.error("RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e) - def setup_amq_channel(self): + async def setup_amq_channel(self, loop): """ Set up the RabbitMQ connection :return: nothing, it applies changes to instance variables """ - credentials: pika.credentials.PlainCredentials = pika.credentials.PlainCredentials( - username=self.amq_configuration.dispatch.rabbit_mq_user, - password=self.amq_configuration.dispatch.rabbit_mq_password - ) - factory = pika.ConnectionParameters( + self.connection = await aio_pika.connect_robust( host=self.amq_configuration.dispatch.amq_host, - credentials=credentials + port=self.amq_configuration.dispatch.amq_port, + login=self.amq_configuration.dispatch.rabbit_mq_user, + password=self.amq_configuration.dispatch.rabbit_mq_password, + loop=loop ) - self.connection = pika.BlockingConnection(factory) - logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_open) - self.connection.add_on_connection_blocked_callback(self.blocked_listener) - self.connection.add_on_connection_unblocked_callback(self.unblocked_listener) - self.channel = self.connection.channel() - self.channel.basic_qos(prefetch_count=self.PREFETCH_COUNT) - if self.amq_configuration.dispatch.use_confirms: - self.setup_local_ack() - self.channel.add_on_return_callback(self.return_listener) + logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_closed) + self.channel = await self.connection.channel() + await self.channel.set_qos(prefetch_count=self.PREFETCH_COUNT) + await self.router.init_routing_paths(self.channel) + self.router.set_route_added_notifier( + lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s", + route.exchange)) - def send_message(self, message: AMQMessage): + async def send_message(self, message: AMQMessage): """ The API routine that sends a datagram message (AMQMessage) to the destination. The Destination is determined from the AMQMessage metadata section, using `domain` and `path` as key. @@ -87,27 +63,31 @@ class RabbitMQProducer: """ try: route = self.router.find_route_by_message(message) - if route: - if route.timeout < 0: + if route and self.rpc_exchange: + if route.timeout <= 0: # No response expected - self.channel.basic_publish( - route.exchange, - self.router.get_routing_key(message), # context path is a routing key - properties=pika.BasicProperties(delivery_mode=2), - body=AMQMessageFactory.serialize(message) + pika_message: aio_pika.message.Message = aio_pika.message.Message( + body=AMQMessageFactory.serialize(message), + content_encoding='utf-8', + delivery_mode=2 + ) + await self.rpc_exchange.publish( + message=pika_message, + routing_key=self.router.get_routing_key(message), # context path is a routing key ) logging.info(f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}") else: # This is RPC call, there should be response - props = pika.BasicProperties( - correlation_id=message.id, + pika_message: aio_pika.message.Message = aio_pika.message.Message( + body=AMQMessageFactory.serialize(message), + content_encoding='utf-8', + delivery_mode=2, + correlation_id=str(message.id), reply_to=self.router.get_reply_to_queue_name() ) - self.channel.basic_publish( - route.exchange, - self.router.get_routing_key(message), # context path is a routing key - properties=props, - body=AMQMessageFactory.serialize(message) + await self.rpc_exchange.publish( + message=pika_message, + routing_key=self.router.get_routing_key(message), # context path is a routing key ) logging.info(" [x] basicPublish (RPC call), messageId: %s, to: %s", message.id.as_string(), route.as_string()) @@ -117,30 +97,8 @@ class RabbitMQProducer: except Exception as err: logging.error("Send message failed: %s", err) - def request_routes(self): - self.router.request_routes_from_adapters() - - def setup_local_ack(self): - self.channel.confirm_delivery() + async def request_routes(self): + await self.router.request_routes_from_adapters() def find_route(self, domain: str, path: str) -> AMQRoute: return self.router.find_route(domain, path) - - def return_listener(self, return_message): - logging.error( - f" [E] ****** (NOT DELIVERED) ID={return_message.properties.message_id}, " - f"Exchg=[{return_message.exchange}], Key=[{return_message.routing_key}], " - f"replyTxt={return_message.reply_text}, length={len(return_message.body) if return_message.body else 0}" - ) - - def blocked_listener(self, reason: str): - logging.error(f" [E] CHANNEL BLOCKED, reason={reason}") - - def unblocked_listener(self): - logging.info(" [*] CHANNEL UNBLOCKED") - - def ack_nack_callback(method): - if method == pika.spec.Basic.Ack: - pass - if method == pika.spec.Basic.Nack: - pass \ No newline at end of file diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py index 251d281..c4587ee 100644 --- a/amqp/router/router_consumer.py +++ b/amqp/router/router_consumer.py @@ -1,9 +1,7 @@ import logging from typing import Set -from pika import spec -from pika.channel import Channel - +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 @@ -30,25 +28,23 @@ class RouterConsumer(RouterProducer): self.amqRouteFactory = AMQRouteFactory() logging.info(" [*] RouterConsumer.") - def init_routing_paths_consumer(self, _channel: Channel): + async def init_routing_paths_consumer(self, _channel: AbstractRobustChannel): """ * Initialize router according to Consumer mode. """ - self.init_routing_paths_consumer(_channel) + await self.init_routing_paths_consumer(_channel) logging.info(" [*] RouterConsumer.initConsumerRouter, channel open: %s", self.is_open(_channel)) if self.is_open(_channel): try: # 1. declare a Queue that gives the RoutingDAta requests cm_request_queue: str = self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_REQ_QUEUE - self.channel.queue_declare(cm_request_queue, durable=True, exclusive=False, auto_delete=False) + queue: AbstractRobustQueue = await self.channel.declare_queue(name=cm_request_queue, durable=True, exclusive=False, auto_delete=False) # 1a. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key) - self.channel.queue_bind(queue=cm_request_queue, exchange=CM_REQUEST_EXCHANGE, 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) - self.channel.basic_consume(queue=cm_request_queue, - auto_ack=False, - on_message_callback=self.handle_routing_data_request_message) + 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: @@ -58,39 +54,35 @@ class RouterConsumer(RouterProducer): * ServiceMessageType.ROUTING_DATA_REQUEST handler listening on queue 'CleverMicroRequest' """ - def handle_routing_data_request_message(self, - cb_channel: Channel, - method: spec.Basic.Deliver, - properties: spec.BasicProperties, - body: bytes): + async def handle_routing_data_request_message(self, message: AbstractIncomingMessage): # some debug statements, TODO - remove for prod release - delivery_tag = method.delivery_tag - debug = "Delivery.Properties = " + properties.__str__() + delivery_tag = message.delivery_tag + debug = "Delivery.Properties = " + message.properties.__str__() logging.info(" [DBG] ******[%] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s", - delivery_tag, method.consumer_tag, method.get_properties(), debug) + delivery_tag, message.consumer_tag, message.properties, debug) - cb_channel.basic_ack(delivery_tag, False) + await message.ack(False) - message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(body) + 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(message), route_mapping) + 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 message.is_valid() and message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping: + if cm_message.is_valid() and cm_message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping: try: service_message: CleverMicroServiceMessage = self.service_message_factory.of( - ServiceMessageType.ROUTING_DATA, route_mapping, message.reply_to + ServiceMessageType.ROUTING_DATA, route_mapping, cm_message.reply_to ) - self.publish_service_message(service_message, CM_RESPONSE_EXCHANGE) + 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.error(" [E] ******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.", - delivery_tag, str(body)) + delivery_tag, str(message.body)) - def register_route(self, route: AMQRoute): + 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. @@ -101,7 +93,7 @@ class RouterConsumer(RouterProducer): serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of( ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT ) - self.publish_service_message(serviceMessage, CM_RESPONSE_EXCHANGE) + await self.publish_service_message(serviceMessage, self.cm_response_exchange) self.add_own_route(route) logging.info(" [DBG] RouterConsumer registered Route: %s", route) else: @@ -110,7 +102,7 @@ class RouterConsumer(RouterProducer): except Exception as ioe: logging.error(" [E] RouterConsumer failed register route data: %s", ioe) - def remove_route_on_cleanup(self, route: AMQRoute) -> bool: + async def remove_route_on_cleanup(self, route: AMQRoute) -> bool: """ * Unregister / remove route from the list * :param route: route to remove @@ -123,7 +115,7 @@ class RouterConsumer(RouterProducer): route.as_string(), ANY_RECIPIENT ) - if not self.publish_service_message(service_message, CM_RESPONSE_EXCHANGE): + 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) diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py index 6503cb8..251be28 100644 --- a/amqp/router/router_producer.py +++ b/amqp/router/router_producer.py @@ -7,10 +7,10 @@ * forwards it to the application. """ import logging -import os -import pika -from pika import BasicProperties, spec -from pika.channel import Channel + +from aio_pika import Message +from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractRobustConnection, AbstractRobustExchange, \ + AbstractIncomingMessage from pika.exchange_type import ExchangeType from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory @@ -29,26 +29,18 @@ class RouterProducer(RouterBase): * :param configuration: adapter's configuration reference. """ super().__init__() - self.connection = None + self.connection: AbstractRobustConnection | None = None self.service_message_factory = CleverMicroServiceMessageFactory(configuration.amq_adapter.service_name) self.amq_configuration: AMQConfiguration = configuration - self.channel: Channel | None = None + self.channel: AbstractRobustChannel | None = None + self.cm_request_exchange: AbstractRobustExchange | None = None + self.cm_response_exchange: AbstractRobustExchange | None = None logging.info(f" [*] RouterProducer.: %s, route: %s", self.amq_configuration.amq_adapter.service_name, self.amq_configuration.amq_adapter.route_mapping) - def amq_consumer(self): - credentials = pika.PlainCredentials( - os.environ['RABBIT_MQ_USER'], os.environ['RABBIT_MQ_PASSWORD'] - ) - self.connection: pika.SelectConnection = pika.SelectConnection( - pika.ConnectionParameters(host=self.amq_configuration.dispatch.amq_host, credentials=credentials) - ) - self.channel = self.connection.channel() - # self.channel.basic_consume() - - def init_routing_paths(self, channel: Channel | None): + async def init_routing_paths(self, channel: AbstractRobustChannel | None): """ * Initialize router according to Producer mode. * The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues @@ -63,22 +55,25 @@ class RouterProducer(RouterBase): try: # **** set up the service discovery internal exchanges **** # 1. declare RequestExchange to where to publish the RoutingData requests - self.channel.exchange_declare(exchange=CM_REQUEST_EXCHANGE, exchange_type=ExchangeType.fanout) + self.cm_request_exchange = await self.channel.declare_exchange( + name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout + ) # # 2. declare ResponseExchange where to publish current RoutingData (response to # a RoutingData request) - self.channel.exchange_declare(exchange=CM_RESPONSE_EXCHANGE, exchange_type=ExchangeType.fanout) + self.cm_response_exchange = await self.channel.declare_exchange( + name=CM_RESPONSE_EXCHANGE, type=ExchangeType.fanout + ) # 2a. declare a Response queue for Routing Data replies from Services cm_response_queue: str = self.get_response_queue_name() - self.channel.queue_declare(cm_response_queue, durable=True, exclusive=False, auto_delete=False, - arguments={}) + response_queue: AbstractRobustQueue = await self.channel.declare_queue( + name=cm_response_queue, durable=True, exclusive=False, auto_delete=False, + arguments={} + ) # 2b. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key) - self.channel.queue_bind(queue=cm_response_queue, exchange=CM_RESPONSE_EXCHANGE, routing_key="#") + await response_queue.bind(exchange=CM_RESPONSE_EXCHANGE, routing_key="#") # 2c. listen for incoming RoutingData messages - self.channel.basic_consume( - queue=cm_response_queue, - auto_ack=False, - on_message_callback=self.handle_routing_data_message) + await response_queue.consume(callback=self.handle_routing_data_message, no_ack=False) logging.info(" [*] Routing master listens for Route updates on %s", cm_response_queue) except Exception as ioe: logging.error(" [E] RouterProducer FATAL: cannot initialize, %s", ioe) @@ -86,38 +81,34 @@ class RouterProducer(RouterBase): else: logging.error(" [E] RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.") - def handle_routing_data_message(self, - cb_channel: Channel, - method: spec.Basic.Deliver, - properties: spec.BasicProperties, - body: bytes): + async def handle_routing_data_message(self, message: AbstractIncomingMessage): """ * Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of * routing data, invokes addRoute or removeRoute depending on the type of the message. """ - logging.info(f" [x] Received {body}, m={method}, p={properties}") + logging.info(f" [x] Received {message.body}, m={message.message_id}, p={message.properties}") """ logging.info(" [DBG] ******[%s] ROUTING_DATA %s >>> %s", delivery.getEnvelope().getDeliveryTag(), delivery.getEnvelope().toString(), str(delivery.getBody())) """ # ACK message now so that it is not being redelivered - cb_channel.basic_ack(method.delivery_tag, False) - message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(body) + await message.ack(multiple=False) + cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body) # ignore the message if it comes from ourselves (even if from different instance) logging.info(" [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s", - method.delivery_tag, - self.service_message_factory.to_string(message), + message.delivery_tag, + self.service_message_factory.to_string(cm_message), self.amq_configuration.amq_adapter.route_mapping, self.amq_configuration.amq_adapter.service_name, - message.reply_to) - if message.is_valid() and not self.amq_configuration.amq_adapter.service_name == message.reply_to: - if message.message_type == ServiceMessageType.ROUTING_DATA: - self.add_routes(message.message) - if message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE: - self.remove_routes(message.message) + cm_message.reply_to) + if cm_message.is_valid() and not self.amq_configuration.amq_adapter.service_name == cm_message.reply_to: + if cm_message.message_type == ServiceMessageType.ROUTING_DATA: + self.add_routes(cm_message.message) + if cm_message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE: + self.remove_routes(cm_message.message) - def request_routes_from_adapters(self): + async def request_routes_from_adapters(self): """ * Send a broadcast message prompting all listening adapters to reply with own routing data. """ @@ -126,37 +117,37 @@ class RouterProducer(RouterBase): serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of( ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT ) - if not self.publish_service_message(serviceMessage, CM_REQUEST_EXCHANGE): + if not await self.publish_service_message(serviceMessage, self.cm_request_exchange): logging.warning(" [E] RouterProducer could not request route data: %s channel is not open.", CM_REQUEST_EXCHANGE) except Exception as ioe: logging.error(" [E] RouterProducer failed request routing data: %s", ioe) - BASIC: BasicProperties = BasicProperties(content_type="application/octet-stream", - content_encoding=None, - headers=None, - delivery_mode=1, - priority=0, - correlation_id=None) - - def publish_service_message(self, message: CleverMicroServiceMessage, exchange_name: str) -> bool: + async def publish_service_message(self, + message: CleverMicroServiceMessage, + exchange: AbstractRobustExchange) -> bool: """ * Publishes a service message to the service queue and logs success log entry. * * :param message: Service Message to be sent - * :param exchangeName: target exchange name where to publish the message - * :param queueName target: queue name (Optional, purpose is to get the queue size only) + * :param exchange: target exchange name where to publish the message * :return True if channel is open, False otherwise * @throws IOException if something else goes wrong """ if self.is_open(self.channel): binary_content: bytes = self.service_message_factory.serialize(message) - self.channel.basic_publish(exchange=exchange_name, - routing_key="", # empty routing key - properties=self.BASIC, - body=binary_content) - logging.info(" [x] Service Message Sent to %s, msg: %s", exchange_name, str(binary_content)) + pika_message: Message = Message( + body=binary_content, + content_encoding='utf-8', + delivery_mode=2, + content_type="application/octet-stream", + headers=None, + priority=0, + correlation_id=None + ) + await exchange.publish(message=pika_message, routing_key="") + logging.info(" [x] Service Message Sent to %s, msg: %s", exchange.name, str(binary_content)) return True return False @@ -181,10 +172,10 @@ class RouterProducer(RouterBase): """ return self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_AMQ_QUEUE - def is_open(self, _channel: Channel) -> bool: + def is_open(self, _channel: AbstractRobustChannel) -> bool: """ * Test if the channel is usable (opened). * :param _channel: the channel * :return True if the channel is opened. """ - return _channel is not None and _channel.is_open + return _channel is not None and not _channel.is_closed diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index 06c823a..14b32cf 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -1,8 +1,10 @@ import logging import logging.config import os +import asyncio from asyncio import Future +from aio_pika.abc import AbstractRobustExchange from amqp.adapter.amq_message_factory import AMQMessageFactory from amqp.adapter.amq_route_factory import AMQRouteFactory @@ -29,53 +31,66 @@ 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: AMQConfiguration, http_adapter): + def __init__(self, amq_configuration: AMQConfiguration, service_adapter): logging.info("***********************************************") logging.info("********** AMQ Service *******************") logging.info("***********************************************") + loop = asyncio.get_event_loop() 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.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name) - self.amq_service_message_handler = AMQServiceMessageHandler(self.http_adapter, self.tracer, self.message_factory) + self.amq_service_message_handler = None self.service_message_factory = CleverMicroServiceMessageFactory( self.amq_configuration.amq_adapter.service_name ) - self.register_routes() + # self.init(loop) will initialize RabbitMQ connection + loop.run_until_complete(self.init(loop, service_adapter)) + # loop.run_until_complete(consume(loop)) # self.consumer_thread = asyncio.create_task(self.rabbit_mq_consumer.channel.start_consuming()) # self.rabbit_mq_consumer.request_routes() logging.info("***********************************************") logging.info("****** AMQ Service Init Complete *********") logging.info("***********************************************") + loop.run_forever() + + async def init(self, loop, service_adapter): + exchange: AbstractRobustExchange = await self.rabbit_mq_consumer.init(loop) + self.amq_service_message_handler = AMQServiceMessageHandler( + service_adapter, self.tracer, self.message_factory, exchange + ) + await self.register_routes() + await 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: + async def reinitialize_if_disconnected(self): + if not self.rabbit_mq_consumer.channel or self.rabbit_mq_consumer.channel.is_closed: self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration) - self.register_routes() - self.rabbit_mq_consumer.request_routes() + await self.rabbit_mq_consumer.init(loop=None) + await self.register_routes() + await self.rabbit_mq_consumer.request_routes() - def register_routes(self): + async def register_routes(self) -> AbstractRobustExchange | None: route_mapping = self.amq_configuration.amq_adapter.route_mapping if AMQRouteFactory.validate(route_mapping): try: - self.rabbit_mq_consumer.register_inbound_routes( - self.amq_service_message_handler.inbound_message_callback_http + await self.rabbit_mq_consumer.register_inbound_routes( + route_mapping, self.amq_service_message_handler.inbound_message_callback ) - self.rabbit_mq_consumer.register_rpc_handler( + return await self.rabbit_mq_consumer.register_rpc_handler( self.amq_service_message_handler.reply_received_callback ) except Exception as e: raise RuntimeError(e) elif route_mapping: logging.info(" [W] route [%s] failed validation, no routes were registered.", route_mapping) + return None def send_message(self, message: AMQMessage, future: Future): self.amq_service_message_handler.outstanding[str(message.id)] = future diff --git a/amqp/service/service_message_handler.py b/amqp/service/service_message_handler.py index 23a4328..c5fe81f 100644 --- a/amqp/service/service_message_handler.py +++ b/amqp/service/service_message_handler.py @@ -1,23 +1,28 @@ import asyncio import logging import time +import concurrent.futures +import os from asyncio import Future from typing import Dict -import pika +from aio_pika import IncomingMessage, Message +from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange from opentelemetry import trace from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator from opentelemetry.trace import Tracer, TraceState -from pika.channel import Channel from amqp.adapter.amq_message_factory import AMQMessageFactory from amqp.adapter.amq_response_factory import AMQResponseFactory from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter from amqp.model.model import AMQResponse, AMQMessage -from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE from amqp.router.serializer import map_as_string +parallel_workers = os.getenv("PARALLEL_WORKERS", default=10) + +# Shared thread pool for parallel execution +executor = concurrent.futures.ThreadPoolExecutor(max_workers=parallel_workers) def collect_trace_info(): trace_map = {} @@ -37,37 +42,32 @@ def set_text(carrier, key, value): class AMQServiceMessageHandler: - def __init__(self, service_adapter: CleverThisServiceAdapter, tracer: Tracer, message_factory: AMQMessageFactory): - self.http_adapter: CleverThisServiceAdapter = service_adapter + def __init__(self, service_adapter: CleverThisServiceAdapter, + tracer: Tracer, message_factory: AMQMessageFactory, + reply_to_exchange: AbstractRobustExchange): + self.service_adapter: CleverThisServiceAdapter = service_adapter self.tracer = tracer self.message_factory: AMQMessageFactory = message_factory + self.reply_to_exchange: AbstractRobustExchange = reply_to_exchange # Dictionary to store outstanding Futures self.outstanding: Dict[str, asyncio.Future] = {} - def inbound_message_callback_http(self, - channel: Channel, - method: pika.spec.Basic.Deliver, - properties: pika.spec.BasicProperties, - body: bytes): + async def inbound_message_callback(self, message: AbstractIncomingMessage): """ Receives the message as bytes from RabbitMQ queue and deserializes it to the AMQMessage Ensures Jaeger trace-id propagation. - In loose coupling mode, invokes appropriate adapter (in v0.1 only HTTPAdapter is supported). - In tight coupling mode, invokes custom handler/mapper/adapter method that - :param body: - :param properties: - :param method: - :param channel: + Invokes custom handler/mapper/adapter method that corresponds to this message + :param message: :return: """ start = time.time() - delivery_tag = method.delivery_tag - debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {method}, props: {properties}" + delivery_tag = message.delivery_tag + debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {message.headers}, props: {message.properties}" logging.info(debug) - amq_message: AMQMessage = AMQMessageFactory.from_bytes(body) + amq_message: AMQMessage = AMQMessageFactory.from_bytes(message.body) logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s", - delivery_tag, method.consumer_tag, amq_message.id, map_as_string(amq_message.trace_info)) + delivery_tag, message.consumer_tag, amq_message.id, map_as_string(amq_message.trace_info)) propagator = TraceContextTextMapPropagator() context = propagator.extract(carrier=amq_message.trace_info, context=trace.context_api.get_current()) @@ -82,21 +82,25 @@ class AMQServiceMessageHandler: # map_as_string(amq_message.trace_info), context_with_span, context) if amq_message.magic == AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST: - response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message)) - self.send_reply(properties.reply_to, delivery_tag, channel, response) + future = executor.submit(self.service_adapter.sync_on_message, amq_message) + # response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message)) + # Wait for the result and send it back to the reply queue + try: + response: AMQResponse = future.result() # Block until the result is available + await self.send_reply(message.reply_to, delivery_tag, response) + except Exception as e: + print(f"[E] Error processing message: {e}") + await message.nack(requeue=False) + return else: logging.error(" [E] Unknown MAGIC value: [%s], don't know how to interpret the message", amq_message.magic) + await message.nack(requeue=False) + return duration = time.time() - start logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration) - channel.basic_ack(delivery_tag, False) - # if "___NACK___" in str(amq_message.body()): - # logging.info(" [x] ######[%s] Done, single NACK: %sms.", delivery_tag, duration) - # channel.basic_nack(delivery_tag, False, False) - # else: - # logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration) - # channel.basic_ack(delivery_tag, False) + await message.ack(multiple=False) def on_http_response(self, delivery, deliveryTag, channel, http_response): """ @@ -119,54 +123,45 @@ class AMQServiceMessageHandler: collect_trace_info() ) try: - self.send_reply(delivery.properties.reply_to, deliveryTag, channel, response) + self.send_reply(delivery.properties.reply_to, deliveryTag, response) except Exception as e: raise RuntimeError(e) - def send_reply(self, reply_to: str, delivery_tag: str, channel: Channel, response: AMQResponse): + async def send_reply(self, reply_to: str, delivery_tag: int | None, response: AMQResponse): """ The Service side code. When the CleverXXX service has output ready, in AMQResponse object, call this method to return the reply back to the caller :param reply_to: :param delivery_tag: - :param channel: :param response: :return: """ if reply_to: logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s", delivery_tag, reply_to, response.id) - reply_props = pika.BasicProperties( + pika_message: Message = Message( + body=AMQResponseFactory.serialize(response), correlation_id=response.id, content_type=response.content_type[0] if isinstance(response.content_type, list) else response.content_type ) - channel.basic_publish( - exchange=AMQ_REPLY_TO_EXCHANGE, - routing_key=reply_to, - body=AMQResponseFactory.serialize(response), - properties=reply_props, + await self.reply_to_exchange.publish( + message=pika_message, + routing_key=reply_to ) - def reply_received_callback(self, - channel: Channel, - method: pika.spec.Basic.Deliver, - properties: pika.spec.BasicProperties, - body: bytes): + async def reply_received_callback(self, message: IncomingMessage): """ The producer/client (API Gateway) facing response_message recipient - must match the received response with the original request and produce the HTTP reply. Note: this code runs on the API Gateway side of RabbitMQ - :param body: - :param properties: - :param method: - :param channel: + :param message: :return: """ - reply_id = properties.correlation_id - content_type = properties.content_type - delivery_tag = method.delivery_tag - debug = f" [DBG] ######[{delivery_tag}] (RPC REPLY) env: {method}, props: {properties}, BODY: {body.decode()}" + reply_id = message.correlation_id + content_type = message.content_type + delivery_tag = message.delivery_tag + debug = f" [DBG] ######[{delivery_tag}] (RPC REPLY) env: {message.headers}, props: {message.properties}, BODY: {message.body.decode()}" logging.info(debug) # TODO - figure out implementation of this part, as this is sending reply back either to a service, # or to the user @@ -184,8 +179,8 @@ class AMQServiceMessageHandler: # internal, service-2-service communication, and the proper response type will be decided than # reply.response().set_result(Response.ok(result.body, content_type).build()) if future: - response: AMQResponse = AMQResponseFactory.from_bytes(body) + response: AMQResponse = AMQResponseFactory.from_bytes(message.body) # Complete the Future with the response data future.set_result(response.body()) - channel.basic_ack(delivery_tag, False) + await message.ack(multiple=False) diff --git a/amqp/service/tracing.py b/amqp/service/tracing.py index 9e1d88f..d54a54d 100644 --- a/amqp/service/tracing.py +++ b/amqp/service/tracing.py @@ -1,5 +1,4 @@ from opentelemetry import metrics, trace -#from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.trace import TracerProvider, Tracer from opentelemetry.sdk.trace.export import BatchSpanProcessor diff --git a/pyproject.toml b/pyproject.toml index 124e2bd..cda395f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.4" +version = "0.2.5" description = "The Python implementation of the AMQ Adaptor for CleverMicro" readme = "README.md" license = { text = "Apache License 2.0" } diff --git a/requirements.txt b/requirements.txt index 3341347..0745550 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,6 @@ aiohttp~=3.11.11 fastapi==0.115.6 -pika~=1.3.2 +aio_pika~=9.5.5 opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp diff --git a/setup.py b/setup.py index df6c0be..8a029ce 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import setup, find_packages setup( name="amqp", - version="0.2.4", + version="0.2.5", author="Stanislav Hejny", author_email="stanislav.hejny@cleverthis.com", description="Adapter for RabbitMQ to integrate a CleverThis python-code Service with CleverMicro platform", -- 2.52.0 From a60c724ae130915533765a71d2c16cf3f5505ee4 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Thu, 13 Mar 2025 16:31:24 +0000 Subject: [PATCH 6/7] Make everythong asynch, prepare for streaming uplpaded file --- amqp/adapter/cleverthis_service_adapter.py | 42 +++--------------- amqp/model/uploaded_file.py | 21 +++++++++ amqp/service/AMQ_PyAdapter.log | 0 amqp/service/multipart_parser.py | 50 ++++++++++++++++++++++ amqp/service/stream_parser.py | 47 ++++++++++++++++++++ 5 files changed, 123 insertions(+), 37 deletions(-) create mode 100644 amqp/model/uploaded_file.py delete mode 100644 amqp/service/AMQ_PyAdapter.log create mode 100644 amqp/service/multipart_parser.py create mode 100644 amqp/service/stream_parser.py diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 693aab1..13d0ad5 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -1,21 +1,18 @@ import asyncio import json import logging -import python_multipart -import io import os import re import threading from aiohttp import ClientSession, ClientResponse -from fastapi import HTTPException, UploadFile -from python_multipart import multipart -from python_multipart.multipart import Field, File +from fastapi import HTTPException from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any from urllib.parse import parse_qs from xml.etree import ElementTree from amqp.adapter.amq_response_factory import AMQResponseFactory +from amqp.service.multipart_parser import CleverMultiPartParser from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQMessage, AMQResponse @@ -25,37 +22,6 @@ lock = threading.Lock() parallel_workers = os.getenv("PARALLEL_WORKERS", default=10) -class CleverMultiPartParser: - def __init__(self, message: AMQMessage): - self.form_data = {} - # Convert each list of strings to a single string joined by newline, then to bytes - headers = {key: "\n".join(value).encode("utf-8") for key, value in message.headers.items()} - content_type: str | bytes | None = headers.get("Content-Type") - content_type, params = multipart.parse_options_header(content_type) - if content_type.decode('utf-8') in ["application/octet-stream", - "multipart/form-data", - "application/x-www-form-urlencoded", - "application/x-url-encoded"]: - try: - python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file) - except Exception as e: - logging.error(f"Error parsing multipart/form-data: {e}") - self.form_data = message.body() - else: - self.form_data = message.body() - - def on_field(self, field: Field): - print(field) - self.form_data[field.field_name.decode('utf-8')] = field.value - print(self.form_data) - - def on_file(self, file: File): - print(file) - self.form_data[file.field_name.decode('utf-8')] = \ - UploadFile(file.file_object, size=file.size, filename=file.file_name) - print(self.form_data) - - def parse_request_body(message: AMQMessage): """ Parses the HTTP POST request body based on the MIME type. @@ -294,6 +260,7 @@ class CleverThisServiceAdapter: :return: AMQResponse to be sent back to the API Gateway via AMQP """ return await self.handle_possible_form( + message, self.session.post( f"http://{self.service_host}:{self.service_port}{message.path}", data=message.body, @@ -302,7 +269,8 @@ class CleverThisServiceAdapter: auth_user ) - async def handle_possible_form(self, request_coroutine, auth_user: Any) -> AMQResponse: + async def handle_possible_form(self, message: AMQMessage, request_coroutine, auth_user: Any) -> AMQResponse: + args = parse_request_body(message) return await request_coroutine async def head(self, message: AMQMessage, auth_user: Any) -> AMQResponse: diff --git a/amqp/model/uploaded_file.py b/amqp/model/uploaded_file.py new file mode 100644 index 0000000..4d14346 --- /dev/null +++ b/amqp/model/uploaded_file.py @@ -0,0 +1,21 @@ +import io + + +class UploadedFile: + def __init__(self, filename: str = None): + self.filename = filename + self.body_b64 = io.BytesIO() + + +def from_bytes(input_bytes: bytes) -> UploadedFile: + """ + Builds the UploadedFile from its serialized (byte array) form. + :param input_bytes: serialized message + :return: UploadedFile instance + """ + prolog_len = (input_bytes[0] << 8) + input_bytes[1] + metadata = input_bytes[2:prolog_len + 2].decode() + body_b64 = input_bytes[prolog_len + 2:] + + return UploadedFile() + diff --git a/amqp/service/AMQ_PyAdapter.log b/amqp/service/AMQ_PyAdapter.log deleted file mode 100644 index e69de29..0000000 diff --git a/amqp/service/multipart_parser.py b/amqp/service/multipart_parser.py new file mode 100644 index 0000000..52406db --- /dev/null +++ b/amqp/service/multipart_parser.py @@ -0,0 +1,50 @@ +import io +import json +import logging + +import python_multipart +from fastapi import UploadFile +from python_multipart import multipart +from python_multipart.multipart import Field, File + +from amqp.model.model import AMQMessage + + +class CleverMultiPartParser: + def __init__(self, message: AMQMessage): + self.form_data = {} + self.is_multipart = False + self.is_json = False + self.is_valid = True + # Convert each list of strings to a single string joined by newline, then to bytes + headers = {key: "\n".join(value).encode("utf-8") for key, value in message.headers.items()} + content_type: str | bytes | None = headers.get("Content-Type") + content_type, params = multipart.parse_options_header(content_type) + if content_type.decode('utf-8') in ["application/octet-stream", + "multipart/form-data", + "application/x-www-form-urlencoded", + "application/x-url-encoded"]: + try: + python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file) + self.is_multipart = True + except Exception as e: + logging.error(f"Error parsing multipart/form-data: {e}") + try: + self.form_data = json.loads(message.body()) + self.is_json = True + except Exception as jsonerror: + logging.error(f"Error parsing JSON: {jsonerror}") + self.is_valid = False + else: + self.form_data = message.body() + + def on_field(self, field: Field): + print(field) + self.form_data[field.field_name.decode('utf-8')] = field.value + print(self.form_data) + + def on_file(self, file: File): + print(file) + self.form_data[file.field_name.decode('utf-8')] = \ + UploadFile(file.file_object, size=file.size, filename=file.file_name) + print(self.form_data) diff --git a/amqp/service/stream_parser.py b/amqp/service/stream_parser.py new file mode 100644 index 0000000..6997781 --- /dev/null +++ b/amqp/service/stream_parser.py @@ -0,0 +1,47 @@ +import json +import struct +import io +import pika + +class StreamParser: + def __init__(self): + self.buffer = io.BytesIO() # Buffer to accumulate chunks + self.metadata_length = None # Length of the metadata + self.metadata = None # Parsed JSON metadata + self.file_writer = None # File writer for the remaining data + + def process_chunk(self, chunk): + """ + Process a chunk of data and handle it according to the rules. + """ + self.buffer.write(chunk) + + # If metadata length is not yet known, try to read it + if self.metadata_length is None and self.buffer.tell() >= 4: + self.buffer.seek(0) + self.metadata_length = struct.unpack('>I', self.buffer.read(4))[0] + + # If metadata length is known but metadata is not yet parsed, try to parse it + if self.metadata_length is not None and self.metadata is None: + if self.buffer.tell() >= 4 + self.metadata_length: + self.buffer.seek(4) + metadata_bytes = self.buffer.read(self.metadata_length) + self.metadata = json.loads(metadata_bytes.decode('utf-8')) + print("Metadata:", self.metadata) + + # Prepare to write the remaining data to a file + self.file_writer = open(self.metadata.get("filename", "output_file"), "wb") + + # If metadata is parsed, write the remaining data to the file + if self.metadata is not None: + remaining_data = self.buffer.read() + if remaining_data: + self.file_writer.write(remaining_data) + + def close(self): + """ + Close the file writer and clean up. + """ + if self.file_writer: + self.file_writer.close() + self.buffer.close() -- 2.52.0 From 4d36700887d481f41a770c9284ab3340f5f8f83e Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Thu, 13 Mar 2025 16:32:11 +0000 Subject: [PATCH 7/7] Make everythong asynch, prepare for streaming uplpaded file --- pyproject.toml | 2 +- setup.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index cda395f..0d9398e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.5" +version = "0.2.6" description = "The Python implementation of the AMQ Adaptor for CleverMicro" readme = "README.md" license = { text = "Apache License 2.0" } diff --git a/setup.py b/setup.py index 8a029ce..6ee9b23 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import setup, find_packages setup( name="amqp", - version="0.2.5", + version="0.2.6", author="Stanislav Hejny", author_email="stanislav.hejny@cleverthis.com", description="Adapter for RabbitMQ to integrate a CleverThis python-code Service with CleverMicro platform", -- 2.52.0