fix(General): Fix deps version so cleverswarm can install it

amqp/adapter/backpressure_handler.py

@@ -4,7 +4,6 @@ import time
 from asyncio import AbstractEventLoop
 from threading import Thread
 
-import psutil
 from aio_pika import Message
 from aio_pika.abc import AbstractRobustChannel
 
@@ -47,40 +46,6 @@ class ScaleRequestV1:
         )
 
 
-class RunningAverage:
-    def __init__(self, num_items):
-        self.buffer = [0] * num_items  # Fixed-size array
-        self.pointer = 0  # Current position in array
-        self.num_items = num_items  # Maximum number of items to store
-        self.is_filled = False  # Track if buffer is fully filled
-
-    def add(self, value):
-        # Store the value at current pointer position
-        self.buffer[self.pointer] = value
-
-        # Move pointer to next position with wrap-around
-        self.pointer += 1
-        if self.pointer >= self.num_items:
-            self.pointer = 0
-            self.is_filled = True
-
-    def get_average(self):
-        # Determine how many items we should average
-        count = self.num_items if self.is_filled else self.pointer
-
-        if count == 0:
-            return 0.0  # Avoid division by zero
-
-        return sum(self.buffer) / count
-
-    def get_current_values(self):
-        """Returns the values in chronological order (oldest first)"""
-        if not self.is_filled:
-            return self.buffer[: self.pointer]
-
-        return self.buffer[self.pointer :] + self.buffer[: self.pointer]
-
-
 class BackpressureHandler:
     # Track the number of messages currently being processed
     current_parallel_executions = 0
@@ -104,7 +69,6 @@ class BackpressureHandler:
         self.swarm_service_id = self.config.amq_adapter.swarm_service_id
         self.swarm_task_id = self.config.amq_adapter.swarm_task_id
         self.parallel_workers = self.config.backpressure.threshold_threads
-        self.average_cpu_usage = None
 
     def increase_parallel_executions(self):
         """Increase the number of parallel executions"""
@@ -155,93 +119,67 @@ class BackpressureHandler:
                 self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
 
     def backpressure_monitor_loop(self):
-        _time_window_millis = self.config.backpressure.time_window
-        _idle_duration_millis = self.config.backpressure.idle_duration
-        _overload_duration_millis = self.config.backpressure.idle_duration
+        _time_window_sec = self.config.backpressure.time_window
+        _idle_duration_millis = 1000 * self.config.backpressure.idle_duration
         _loop = asyncio.new_event_loop()
-        _monitor_interval = 0.2  # Monitor every 200ms
-        CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2
-        IDLE_THRESHOLD, OVERLOAD_THRESHOLD = 15, 90
-        self.average_cpu_usage = RunningAverage(
-            int(
-                max(_overload_duration_millis, _idle_duration_millis)
-                // _monitor_interval
-            )
-        )
 
         async def _backpressure_monitor():
             """Monitor the backpressure conditions and trigger events accordingly"""
-            _cpu_usage_state = CPU_ACTIVE
-            _cpu_usage_changed = time.time()
             while True:
-                _current_cpu_usage = psutil.cpu_percent(interval=None)
-                self.average_cpu_usage.add(_current_cpu_usage)
-                _average_cpu_usage = self.average_cpu_usage.get_average()
-                _now = time.time()
-                # logging_info(
-                #    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
-                #    _current_cpu_usage,
-                #    _average_cpu_usage,
-                #    _cpu_usage_state,
-                #    self.last_data_message_time,
-                #    self.last_backpressure_event,
-                #    self.last_backpressure_event_time,
-                # )
-                if _average_cpu_usage < IDLE_THRESHOLD:
-                    if _cpu_usage_state != CPU_IDLE:
-                        _cpu_usage_changed = _now
-                    _cpu_usage_state = CPU_IDLE
-                elif _average_cpu_usage > OVERLOAD_THRESHOLD:
-                    if _cpu_usage_state != CPU_OVERLOAD:
-                        _cpu_usage_changed = _now
-                    _cpu_usage_state = CPU_OVERLOAD
-                else:
-                    if _cpu_usage_state != CPU_ACTIVE:
-                        _cpu_usage_changed = _now
-                    _cpu_usage_state = CPU_ACTIVE
-
+                logging_info(
+                    "Backpressure: Monitor loop, current=%s",
+                    self.current_parallel_executions,
+                )
+                logging_info(
+                    "Backpressure: Last data message time=%s, eventTime=%s",
+                    self.last_data_message_time,
+                    self.last_backpressure_event_time,
+                )
+                logging_info(
+                    "Backpressure: Last backpressure event=%s",
+                    self.last_backpressure_event,
+                )
                 # Check if the current time exceeds the last backpressure event time
                 if (
-                    _cpu_usage_state == CPU_OVERLOAD
-                    and _now - _cpu_usage_changed > _overload_duration_millis
-                    and (
-                        _now - self.last_backpressure_event_time > _time_window_millis
-                        or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
-                    )
+                    self.current_parallel_executions >= self.parallel_workers
+                    and time.time() - self.last_backpressure_event_time
+                    > _time_window_sec
+                    and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
                 ):
                     # Trigger the overload event
                     await self.handle_backpressure_overload_event()
-                    self.last_backpressure_event_time = _now
+                    self.last_backpressure_event_time = time.time()
                     self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
                 elif (
-                    _cpu_usage_state == CPU_IDLE
-                    and _now - _cpu_usage_changed > _idle_duration_millis
-                    and _now - self.last_data_message_time > _idle_duration_millis
-                    and (
-                        _now - self.last_backpressure_event_time > _time_window_millis
-                        or self.last_backpressure_event != ScalingRequestAlert.IDLE
-                    )
+                    self.current_parallel_executions == 0
+                    and time.time() - self.last_data_message_time
+                    >= _idle_duration_millis
+                    and self.last_backpressure_event != ScalingRequestAlert.IDLE
                 ):
                     # Trigger the idle event
                     await self._handle_backpressure_idle_event()
                     self.last_backpressure_event = ScalingRequestAlert.IDLE
-                    self.last_backpressure_event_time = _now
+                    self.last_backpressure_event_time = time.time()
                 else:
                     # Trigger update event
-                    if _now - self.last_backpressure_event_time > _time_window_millis:
+                    if (
+                        self.current_parallel_executions > 0
+                        and time.time() - self.last_backpressure_event_time
+                        > _time_window_sec
+                    ):
                         # Trigger the update event
                         _scaling_request: ScaleRequestV1 = ScaleRequestV1(
                             self.swarm_service_id,
                             self.swarm_task_id,
-                            100,
-                            _average_cpu_usage,
+                            self.parallel_workers,
+                            self.parallel_workers - self.current_parallel_executions,
                             ScalingRequestAlert.UPDATE,
                         )
                         # Address the message to any adapter capable of supporting BACKPRESSURE request
                         await self.publish_backpressure_request(_scaling_request)
-                        self.last_backpressure_event_time = _now
+                        self.last_backpressure_event_time = time.time()
                         self.last_backpressure_event = ScalingRequestAlert.UPDATE
-                await asyncio.sleep(_monitor_interval)
+                await asyncio.sleep(_time_window_sec)
 
         _loop.run_until_complete(_backpressure_monitor())
 

amqp/adapter/cleverthis_service_adapter.py

@@ -17,16 +17,14 @@ from aio_pika.abc import (
 from aiohttp import ClientSession, ClientResponse
 from dataclasses import dataclass
 from fastapi import HTTPException
-from pydantic import BaseModel
 from starlette.responses import FileResponse
 
-from amqp.adapter import serializer, pydantic_serializer
 from amqp.adapter.data_parser import AMQDataParser
 from amqp.adapter.data_response_factory import DataResponseFactory
 from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
 from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import DataMessage, DataResponse, AMQErrorMessage
+from amqp.model.model import DataMessage, DataResponse
 from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
 from amqp.router.utils import await_result
 
@@ -210,10 +208,8 @@ class CleverThisServiceAdapter:
             else:
                 raise ValueError(f"Unexpected HTTP method: {message.method}")
         else:
-            _amq_response = DataResponseFactory.of_error_message(
-                message.id,
-                AMQErrorMessage(401, "Unauthorized", "Unauthorized"),
-                message.trace_info,
+            _amq_response = DataResponseFactory.of(
+                message.id, 401, "text/plain", b"Unauthorized", message.trace_info
             )
         logging_info(
             f"ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}"
@@ -248,9 +244,11 @@ class CleverThisServiceAdapter:
                 await _http_resp.read(),
                 message.trace_info,
             )
-        return DataResponseFactory.of_error_message(
+        return DataResponseFactory.of(
             message.id,
-            AMQErrorMessage(404, "Not Found", "Destination location does not exists"),
+            404,
+            "text/plain",
+            str("Destination location does not exists").encode("utf-8"),
             message.trace_info,
         )
 
@@ -347,9 +345,7 @@ class CleverThisServiceAdapter:
                     elif arg.annotation == float:
                         _verified_args[arg.name] = float(args[arg.name])
                     elif arg.annotation == bool:
-                        _verified_args[arg.name] = serializer.str_to_bool(
-                            args[arg.name]
-                        )
+                        _verified_args[arg.name] = bool(args[arg.name])
                     # TODO: more?
                     else:
                         _verified_args[arg.name] = args[arg.name]
@@ -375,9 +371,6 @@ class CleverThisServiceAdapter:
                     ),
                     "application/octet-stream",
                 )
-            elif isinstance(_result, BaseModel):
-                _json_result = _result.json()
-                _content_type = "application/json"
             else:
                 _json_result = _result
                 _content_type = "application/json"
@@ -406,18 +399,10 @@ class CleverThisServiceAdapter:
                 message.trace_info,
             )
         except HTTPException as http_error:
-            logging_error(f"Service REST endpoint invocation failed: {http_error}")
             _content = str(http_error.detail)
             if not is_likely_json(_content):
-                return DataResponseFactory.of_error_message(
-                    message.id,
-                    AMQErrorMessage(
-                        http_error.status_code,
-                        "Service Invocation Failed",
-                        _content,
-                    ),
-                    message.trace_info,
-                )
+                _content = '{"error": "' + _content + '"}'
+            logging_error(f"Service REST endpoint invocation failed: {http_error}")
             return DataResponseFactory.of(
                 message.id,
                 http_error.status_code,
@@ -429,15 +414,7 @@ class CleverThisServiceAdapter:
             logging_error(f"Service endpoint invocation failed: {error}")
             _content = str(error)
             if not is_likely_json(_content):
-                return DataResponseFactory.of_error_message(
-                    message.id,
-                    AMQErrorMessage(
-                        500,
-                        "Service Invocation Failed",
-                        _content,
-                    ),
-                    message.trace_info,
-                )
+                _content = '{"error": "' + _content + '"}'
             return DataResponseFactory.of(
                 message.id,
                 500,

amqp/adapter/data_parser.py

@@ -1,7 +1,6 @@
 import io
 import json
 import python_multipart
-from tempfile import SpooledTemporaryFile
 
 from fastapi import UploadFile
 from python_multipart.multipart import Field, File
@@ -71,12 +70,8 @@ class MultipartFormDataParser:
 
     def on_file(self, file: File):
         logging_debug(str(file))
-        file.file_object.seek(0)
-        _spooled_file: SpooledTemporaryFile = SpooledTemporaryFile()
-        _spooled_file.write(file.file_object.read())
-        _spooled_file.seek(0)
         self.form_data[file.field_name.decode("utf-8")] = UploadFile(
-            _spooled_file, size=file.size, filename=file.file_name.decode("utf-8")
+            file.file_object, size=file.size, filename=file.file_name
         )
 
 

amqp/adapter/data_response_factory.py

@@ -5,7 +5,7 @@ method to create, serialize and deserialize DataResponse message
 import base64
 import re
 from io import BytesIO
-from amqp.model.model import DataResponse, AMQErrorMessage
+from amqp.model.model import DataResponse
 from amqp.model.snowflake_id import SnowflakeId
 from amqp.adapter.serializer import parse_map_string, map_as_string
 
@@ -31,8 +31,6 @@ class DataResponseFactory:
         content_type: str,
         body: bytes,
         trace_info: dict[str, str],
-        error: str | None = None,
-        error_cause: str | None = None,
     ) -> DataResponse:
         """
         Create the DataResponse message supplying all key values
@@ -48,27 +46,11 @@ class DataResponseFactory:
             response_code=response_code,
             content_type=content_type,
             response=body,
-            error=error,
-            error_cause=error_cause,
+            error=None,
+            error_cause=None,
             trace_info=trace_info,
         )
 
-    @staticmethod
-    def of_error_message(
-        id: SnowflakeId,
-        error_message: AMQErrorMessage,
-        trace_info: dict[str, str],
-    ) -> DataResponse:
-        return DataResponseFactory.of(
-            id,
-            error_message.response_code,
-            error_message.content_type,
-            error_message.serialize(),
-            trace_info,
-            error=error_message.error,
-            error_cause=error_message.detail,
-        )
-
     @staticmethod
     def serialize(response: DataResponse) -> bytes:
         """

amqp/adapter/logging_utils.py

@@ -4,11 +4,8 @@ import os
 import sys
 import threading
 import traceback
-from importlib.metadata import version
 
-# Get version from installed package metadata
-__version__ = version("amq_adapter")  # Name must match pyproject.toml [project] name
-__verbose__ = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
+verbose = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
 
 
 def get_context_info():
@@ -54,18 +51,18 @@ def logging_error(msg: str, *args) -> None:
     if _exec_info:
         _tb = traceback.extract_tb(sys.exc_info()[2])[-1]
         logging.error(
-            f"{_thread_id}:{__version__} [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
+            f"{_thread_id} [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
             *args,
         )
     else:
         module, line_number, function_name = get_context_info()
         if module and line_number and function_name:
             logging.error(
-                f"{_thread_id}:{__version__} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+                f"{_thread_id} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
                 *args,
             )
         else:
-            logging.error(f"{_thread_id}:{__version__} [E] {msg}", *args)
+            logging.error(f"{_thread_id} [E] {msg}", *args)
 
 
 def logging_warning(msg: str, *args) -> None:
@@ -76,17 +73,17 @@ def logging_warning(msg: str, *args) -> None:
         msg (str): The warning message to log.
     """
     _thread_id = threading.get_ident()
-    if not __verbose__:
-        logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
+    if not verbose:
+        logging.warning(f"{_thread_id} [W] {msg}", *args)
         return
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.warning(
-            f"{_thread_id}:{__version__} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
+        logging.warning(f"{_thread_id} [W] {msg}", *args)
 
 
 def logging_info(msg: str, *args) -> None:
@@ -97,17 +94,17 @@ def logging_info(msg: str, *args) -> None:
         msg (str): The info message to log.
     """
     _thread_id = threading.get_ident()
-    if not __verbose__:
-        logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
+    if not verbose:
+        logging.info(f"{_thread_id} [*] {msg}", *args)
         return
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.info(
-            f"{_thread_id}:{__version__} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
+        logging.info(f"{_thread_id} [*] {msg}", *args)
 
 
 def logging_debug(msg: str, *args) -> None:
@@ -121,8 +118,8 @@ def logging_debug(msg: str, *args) -> None:
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.info(
-            f"{_thread_id}:{__version__} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.debug(f"{_thread_id}:{__version__} [DBG] {msg}", *args)
+        logging.debug(f"{_thread_id} [DBG] {msg}", *args)

amqp/adapter/serializer.py

@@ -63,7 +63,3 @@ def parse_map_list_string(input_str: str) -> Dict[str, List[str]]:
         for token in input_str[1:-1].split("],"):
             add_to_map(map_, token)
     return map_
-
-
-def str_to_bool(value: str) -> bool:
-    return value.lower() in ("true", "yes", "1", "t", "y", "on")

amqp/config/application.properties

@@ -2,41 +2,30 @@
 # CleverMicro AMQ Adapter settings
 # ====================================================================
 [CleverMicro-AMQ]
-# CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters
-
-# A name to identify this service for the AMQ Adapter
-cm.amq-adapter.service-name=cleverthis-service-name
-# The CleverMicro AMQ Adapter instance ID, used to identify this instance in the CleverMicro AMQ Adapter cluster
-cm.amq-adapter.generator-id=1
-# Route mapping. For route detail description see section 3.1 in Onboarding Guide for Python:
-# https://docs.cleverthis.com/en/architecture/microservices/onboarding/python
-# NOTE: this is a comma-separated list of route mappings (multiple, comma-separated routes are allowed here)
-cm.amq-adapter.route-mapping=
-# Indicate if AMQ Adapter should respond with HTTP 401 Unauthorized if the request does not contain valid JWT token
+cm.amq-adapter.service-name=amq-adapter-python
+cm.amq-adapter.generator-id=164
+cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
 cm.amq-adapter.require-authenticated-user=false
 
-# CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding
 cm.dispatch.use-dlq=true
+cm.dispatch.use-confirms=false
 cm.dispatch.amq-host=rabbitmq
 cm.dispatch.amq-port=5672
 cm.dispatch.amq-port-tls=5671
 cm.dispatch.download-dir=/tmp/downloaded_files
-# Not used for CleveSwarm nor CleverBRAG or in tight coupling mode.
+# Not used for CleverSwarm nor CleverBRAG or in tight coupling mode.
 # applicable only in loose coupling mode, specify the destination where to forward the REST request
-#cm.dispatch.service-host=cleverthis-service-name.localhost
+#cm.dispatch.service-host=cleverswarm.localhost
 #cm.dispatch.service-port=8080
-
-# Backpressure monitor settings
+#
 # Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert
 cm.backpressure.threshold=5
-# Define maximum CPU usage (%) before triggering backpressure OVERLOAD alert
-cm.backpressure.threshold-cpu-overload=90
-# Define maximum CPU usage (%) before triggering backpressure IDLE alert
-cm.backpressure.threshold-cpu-idle=10
+# Define maximum CPU usage before triggering backpressure OVERLOAD alert
+cm.backpressure.threshold-cpu=90
 # Define the time window between the Backpressure reports, in seconds
 cm.backpressure.time-window=10
 # Define the time window length, in seconds, during which, if CPU usage is consistently above the threshold, the system will trigger OVERLOAD alert
 cm.backpressure.cpu-overload-duration=30
 # Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert
-cm.backpressure.cpu-idle-duration=30
+cm.backpressure.idle-duration=30
 

amqp/model/model.py

@@ -305,30 +305,3 @@ class ScalingRequest:
                 data["requestType"]
             ),  # Convert string to Enum
         )
-
-
-class AMQErrorMessage:
-    """
-    The error response message to RPC style call (note: the request is made with DataMessage)
-    """
-
-    def __init__(
-        self,
-        response_code: int,
-        error: str,
-        detail: str,
-        content_type: str = "application/json",
-    ):
-        self.response_code = response_code
-        self.error = error
-        self.detail = detail
-        self.content_type = content_type
-
-    def serialize(self) -> bytes:
-        return json.dumps(
-            {
-                "response_code": self.response_code,
-                "error": self.error,
-                "detail": self.detail,
-            }
-        ).encode("utf-8")

amqp/rabbitmq/rabbit_mq_client.py

@@ -1,8 +1,7 @@
 import logging
 
 import aio_pika
-from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
-from pika.exchange_type import ExchangeType
+from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange, ExchangeType
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.logging_utils import (
@@ -55,7 +54,7 @@ class RabbitMQClient:
             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=ExchangeType.direct
+            AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.DIRECT
         )
         await self.reply_to_queue.bind(
             exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue
@@ -114,7 +113,7 @@ class RabbitMQClient:
                     )
                     exchange: AbstractRobustExchange = (
                         await self.channel.declare_exchange(
-                            name=_exchange_name, type=ExchangeType.topic
+                            name=_exchange_name, type=ExchangeType.TOPIC
                         )
                     )
                     await queue.bind(exchange, route.key)

amqp/router/router_producer.py

@@ -15,9 +15,8 @@ from aio_pika.abc import (
     AbstractRobustQueue,
     AbstractRobustConnection,
     AbstractRobustExchange,
-    AbstractIncomingMessage,
+    AbstractIncomingMessage, ExchangeType,
 )
-from pika.exchange_type import ExchangeType
 
 from amqp.adapter.logging_utils import (
     logging_error,
@@ -82,13 +81,13 @@ class RouterProducer(RouterBase):
                 # **** set up the service discovery internal exchanges ****
                 # 1. declare RequestExchange to where to publish the RoutingData requests
                 self.cm_request_exchange = await self.channel.declare_exchange(
-                    name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout
+                    name=CM_REQUEST_EXCHANGE, type=ExchangeType.FANOUT
                 )
                 #
                 # 2. declare ResponseExchange where to publish current RoutingData (response to
                 #    a RoutingData request)
                 self.cm_response_exchange = await self.channel.declare_exchange(
-                    name=CM_RESPONSE_EXCHANGE, type=ExchangeType.fanout
+                    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()

amqp/service/amq_message_handler.py

@@ -75,21 +75,12 @@ class DataMessageHandler:
         self.file_handler: FileHandler = file_handler
         self.backpressure_handler: BackpressureHandler = backpressure_handler
 
-    async def inbound_data_message_callback_as_thread(
-        self, message: AbstractIncomingMessage
-    ):
-        """
-        The callback method that is called when a DataMessage is received from RabbitMQ, starts new Thread.
-        """
+    async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
         threading.Thread(
-            target=lambda: asyncio.run(
-                self.inbound_data_message_callback_no_thread(message)
-            )
+            target=lambda: asyncio.run(self.run_callback_thread(message))
         ).start()
 
-    async def inbound_data_message_callback_no_thread(
-        self, message: AbstractIncomingMessage
-    ):
+    async def run_callback_thread(self, message: AbstractIncomingMessage):
         """
         Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object
         Ensures Jaeger trace-id propagation.

amqp/service/amq_service.py

@@ -107,7 +107,7 @@ class AMQService:
             try:
                 await self.rabbit_mq_client.register_inbound_routes(
                     route_mapping,
-                    self.amq_data_message_handler.inbound_data_message_callback_no_thread,
+                    self.amq_data_message_handler.inbound_data_message_callback,
                 )
                 return await self.rabbit_mq_client.register_data_reply_callback(
                     self.amq_data_message_handler.reply_received_callback

cleverswarm/clever_swarm_adapter.py

@@ -26,7 +26,7 @@ from amqp.adapter.data_parser import AMQDataParser
 from amqp.adapter.data_response_factory import AMQResponseFactory
 from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQResponse, AMQErrorMessage
+from amqp.model.model import AMQResponse
 from amqp.service.amq_service import AMQService
 
 
@@ -220,12 +220,12 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
                     loop=self.loop,
                 )
 
-        return AMQResponseFactory.of_error_message(
+        return AMQResponseFactory.of(
             message.id,
-            AMQErrorMessage(
             404,
-                "Not Found",
-                f"Destination location [{message.path}] does not exists",
+            "text/plain",
+            str(f"Destination location [{message.path}] does not exists").encode(
+                "utf-8"
             ),
             message.trace_info,
         )
@@ -252,13 +252,11 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
                 for _body_param in _route.dependant.body_params:
                     if _body_param.name in _form_data:
                         _file_data = _form_data[_body_param.name]
+                        actual_filepath = message.extra_data[_body_param.name]
                         _form_data[_body_param.name] = (
                             _body_param.type_.__name__ == "UploadFile"
-                            and message.extra_data.get(_body_param.name)
                             and UploadFile(
-                                file=pathlib.Path(
-                                    message.extra_data.get(_body_param.name)
-                                ).open("rb"),
+                                file=pathlib.Path(actual_filepath).open("rb"),
                                 size=_file_data["size"],
                                 filename=_file_data["filename"],
                                 headers=Headers(
@@ -293,12 +291,12 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
                     loop=self.loop,
                 )
 
-        return AMQResponseFactory.of_error_message(
+        return AMQResponseFactory.of(
             message.id,
-            AMQErrorMessage(
             404,
-                "Not Found",
-                f"Destination location [{message.path}] does not exists",
+            "text/plain",
+            str(f"Destination location [{message.path}] does not exists").encode(
+                "utf-8"
             ),
             message.trace_info,
         )

pyproject.toml

@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.30"
-description = "The Python implementation of the AMQ Adapter for CleverMicro"
+version = "0.2.23-dev"
+description = "The Python implementation of the AMQ Adaptor for CleverMicro"
 readme = "README.md"
 authors = [
     { name = "Stanislav Hejny", email = "stanislav.hejny@cleverthis.com" }
@@ -19,16 +19,15 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "aio-pika",
-    "aiohttp",
-    "fastapi",
+    "aio-pika~=9.5",
+    "aiohttp~=3.11",
+    "aio_pika~=9.5",
+    "fastapi~=0.114",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
-    "pika",
-    "psutil",
     "python_multipart"
 ]
 

tools/create_exchange.py

@@ -1,90 +0,0 @@
-import asyncio
-import aio_pika
-import os
-import argparse
-import sys
-
-from aio_pika.abc import AbstractRobustExchange
-
-
-async def create_exchange(loop, rabbit_user, rabbit_password, rabbit_host):
-    """
-    Asynchronously connects to RabbitMQ, declares a direct exchange,
-    and prints the result.  Handles potential connection errors.
-    """
-    exchange_name = "cleverthis.clevermicro.management"
-    connection = None  # Keep track of the connection
-
-    try:
-        # Construct the connection URI from environment variables
-        connection_uri = f"amqp://{rabbit_user}:{rabbit_password}@{rabbit_host}/"
-
-        # Attempt to establish the connection
-        connection = await aio_pika.connect_robust(connection_uri, loop=loop)
-        print(f"Connected to RabbitMQ: {connection_uri}")
-
-        # Create a channel
-        channel = await connection.channel()
-
-        # Declare the exchange
-        exchange: AbstractRobustExchange = await channel.declare_exchange(
-            exchange_name, aio_pika.ExchangeType.DIRECT
-        )
-        print(f"Exchange '{exchange.name}' created successfully.")
-
-    except aio_pika.exceptions.AMQPConnectionError as e:
-        print(f"Error: Could not connect to RabbitMQ.  Details: {e}")
-        sys.exit(1)  # Exit with an error code
-    except aio_pika.exceptions.AMQPChannelError as e:
-        print(f"Error creating channel or exchange: {e}")
-        sys.exit(1)
-    except Exception as e:
-        print(f"An unexpected error occurred: {e}")
-        sys.exit(1)
-    finally:
-        # Ensure the connection is closed, even if errors occur
-        if connection:
-            await connection.close()
-            print("Connection to RabbitMQ closed.")
-
-def main():
-    """
-    Main function to parse command-line arguments, retrieve RabbitMQ
-    credentials from environment variables, and run the
-    async exchange creation.
-    """
-    # Set up argument parsing
-    parser = argparse.ArgumentParser(description="Create a RabbitMQ exchange.")
-    parser.add_argument(
-        "--host",
-        default="127.0.0.1",
-        help="RabbitMQ host (default: localhost)",
-    )
-
-    # Parse the arguments
-    args = parser.parse_args()
-    rabbit_host = args.host
-
-    # Get RabbitMQ credentials from environment variables
-    rabbit_user = os.environ.get("RABBIT_MQ_USER")
-    rabbit_password = os.environ.get("RABBIT_MQ_PASSWORD")
-
-    if not rabbit_user or not rabbit_password:
-        print(
-            "Error: RabbitMQ credentials not found in environment variables.\n"
-            "Please set RABBIT_MQ_USER and RABBIT_MQ_PASSWORD."
-        )
-        sys.exit(1)  # Exit with an error code
-
-    # Create the event loop
-    loop = asyncio.get_event_loop()
-
-    # Run the asynchronous exchange creation
-    loop.run_until_complete(create_exchange(loop, rabbit_user, rabbit_password, rabbit_host))
-
-    # Close the loop
-    loop.close()
-
-
-if __name__ == "__main__":
-    main()