Make everythong asynch, prepare for streaming uplpaded file

amqp/adapter/cleverthis_service_adapter.py

@@ -1,38 +1,25 @@
+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.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
 
-
-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()}
-        python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file)
-
-    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)
+# Track the number of messages currently being processed
+current_parallel_executions = 0
+lock = threading.Lock()
+parallel_workers = os.getenv("PARALLEL_WORKERS", default=10)
 
 
 def parse_request_body(message: AMQMessage):
@@ -170,6 +157,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.
@@ -177,6 +172,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:
@@ -185,22 +188,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  ============================
@@ -253,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,
@@ -261,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:

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()
+

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:
+                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)
-    """

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

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.<init>")
 
-    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)
 

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.<init>: %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

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

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)

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,53 +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
@@ -183,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)

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()

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

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.2"
+version = "0.2.6"
 description = "The Python implementation of the AMQ Adaptor for CleverMicro"
 readme = "README.md"
 license = { text = "Apache License 2.0" }

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

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name="amqp",
-    version="0.2.0",
+    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",