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

.flake8

@@ -1,4 +0,0 @@
-[flake8]
-max-line-length = 88
-ignore = E501, W503, E203
-exclude = .git,__pycache__,docs/source/conf.py,old,build,dist,.venv

.pre-commit-config.yaml

@@ -1,21 +0,0 @@
-repos:
-- repo: https://github.com/pre-commit/pre-commit-hooks
-  rev: v4.2.0
-  hooks:
-  - id: check-yaml
-  - id: end-of-file-fixer
-  - id: trailing-whitespace
-- repo: https://github.com/PyCQA/isort
-  rev: 6.0.1
-  hooks:
-  - id: isort
-    args: ["--profile", "black"]
-- repo: https://github.com/psf/black
-  rev: 25.1.0
-  hooks:
-  - id: black
-    language_version: python3.11
-- repo: https://github.com/PyCQA/flake8
-  rev: 7.2.0
-  hooks:
-  - id: flake8

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

@@ -7,25 +7,24 @@ import asyncio
 import inspect
 import json
 from asyncio import AbstractEventLoop
-from dataclasses import dataclass
-from typing import Any, Callable, Coroutine, Dict, List
+from typing import Dict, List, Callable, Coroutine, Any
 
 from aio_pika.abc import (
-    AbstractExchange,
     AbstractRobustChannel,
     AbstractRobustConnection,
+    AbstractExchange,
 )
-from aiohttp import ClientResponse, ClientSession
+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
+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_debug, logging_error, logging_info
+from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQErrorMessage, DataMessage, DataResponse
+from amqp.model.model import DataMessage, DataResponse
 from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
 from amqp.router.utils import await_result
 
@@ -96,7 +95,7 @@ async def _convert_file_response(
         _exchange, obj.path, _routing_key, loop
     )
     # as above, but luckily this time we don't need to wait for the result
-    asyncio.run_coroutine_threadsafe(_channel.close(), loop)
+    _future = asyncio.run_coroutine_threadsafe(_channel.close(), loop)
     return json.dumps(
         {
             "files": [
@@ -209,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}"
@@ -247,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,
         )
 
@@ -261,7 +260,7 @@ class CleverThisServiceAdapter:
         :param auth_user: user authenticated by the API Gateway, as UserSchem
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        return await self._authorize_forward_request(
+        return await self.handle_possible_form(
             message,
             self.session.put(
                 f"http://{self.service_host}:{self.service_port}{message.path}",
@@ -279,7 +278,7 @@ class CleverThisServiceAdapter:
         :param auth_user: user authenticated by the API Gateway, as UserSchem
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        return await self._authorize_forward_request(
+        return await self.handle_possible_form(
             message,
             self.session.post(
                 f"http://{self.service_host}:{self.service_port}{message.path}",
@@ -289,14 +288,10 @@ class CleverThisServiceAdapter:
             auth_user,
         )
 
-    async def _authorize_forward_request(
+    async def handle_possible_form(
         self, message: DataMessage, request_coroutine, auth_user: Any
     ) -> DataResponse:
-        """
-        This method is used to authorize the request and forward it to the CleverThis Service API.
-        However, it is only used for loose / 3rd party mode of coupling with the Service
-        """
-        # AMQDataParser.parse_request_body(message)
+        _args = AMQDataParser.parse_request_body(message)
         return await request_coroutine
 
     async def head(self, message: DataMessage, auth_user: Any) -> DataResponse:
@@ -350,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]
@@ -378,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"
@@ -409,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,
@@ -432,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/file_handler.py

@@ -4,22 +4,22 @@ implements the file downloader for the amqp adapter, where file is sent in chunk
 
 import asyncio
 import json
-import logging
 import os
-from asyncio import AbstractEventLoop
+import logging
+from asyncio import Future, AbstractEventLoop
 from collections import defaultdict
 from pathlib import Path
 
-from aio_pika import Message, connect_robust
+from aio_pika import connect_robust, Message
 from aio_pika.abc import (
-    AbstractExchange,
     AbstractIncomingMessage,
-    AbstractQueue,
     AbstractRobustChannel,
+    AbstractQueue,
+    AbstractExchange,
     DeliveryMode,
 )
 
-from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
+from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.router.utils import await_result
 
 # Global dictionary to track completed downloads
@@ -166,6 +166,7 @@ class StreamingFileHandler(FileHandler):
             message_data (dict): The entire message
             output_dir (str): The directory where the file should be saved.
         """
+        global download_locations
         filename = file_info["filename"]
         size = file_info["size"]
         stream_name = file_info["streamName"]
@@ -280,6 +281,7 @@ class StreamingFileHandler(FileHandler):
             loop: The asyncio event loop of the RabbitMQ API
             output_dir
         """
+        global download_locations
         try:
             _message_data = json.loads(message.body.decode())
             _amq_message_data = json.loads(json_data.decode())

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/adapter/service_message_factory.py

@@ -1,12 +1,13 @@
 import base64
+import logging
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.logging_utils import logging_error, logging_info
-from amqp.adapter.serializer import map_as_string, parse_map_string
 from amqp.adapter.trace_info_adapter import TraceInfoAdapter
 from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
+from amqp.adapter.serializer import parse_map_string, map_as_string
 
 RS = "|"
 SPLIT_RS = r"\|"

amqp/config/amq_configuration.py

@@ -5,7 +5,7 @@ import os
 from amqp.adapter.logging_utils import logging_error, logging_warning
 
 """
-Convert configuration from properties file to an object.
+Convert configuration from properties file to an object. 
 """
 
 

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

@@ -1,8 +1,8 @@
 import base64
 import json
-from dataclasses import dataclass, field
 from enum import Enum
-from typing import Dict, List
+from typing import List, Dict
+from dataclasses import dataclass, field
 
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
@@ -16,24 +16,9 @@ Define the structure of messages used in Clever Micro
 """
 
 
-@dataclass
 class CleverMicroMessage:
     DEFAULT_CONTENT_TYPE = ["application/octet-stream"]
 
-    """
-    The common structure of a message that transfers the user request to the Service and back. All data related messages
-    conform to this base structure:
-
-    magic - identifies the message type (REST, RPC, DataResponse)
-    id - unique message ID
-    m_field - method name (for REST style calls) or service name (for RPC style calls) or response code for DataResponse
-    d_field - domain name (for REST style calls) or module name (for RPC style calls) or  error code for DataResponse
-    p_field - path (for REST style calls) or method name (for RPC style calls) or error cause for DataResponse
-    headers - headers of the message (e.g. Content-Type, Accept, etc.)
-    trace_info - trace information (e.g. trace ID, span ID, etc. for Jaeger trace monitor)
-    base64body - the body of the message encoded in base64
-    """
-
     def __init__(
         self,
         magic: str,
@@ -81,19 +66,11 @@ class CleverMicroMessage:
     def body(self) -> bytes:
         return base64.b64decode(self.base64body())
 
-    def content_type(self) -> str:
+    def contentType(self) -> str:
         return self.headers().get("Content-Type", self.DEFAULT_CONTENT_TYPE)[0]
 
 
 class DataMessage(CleverMicroMessage):
-    """
-    The class represents a structure of a message that transfers the user request to the Service.
-    It may also be used to make a call/request between services.
-    The response to this call is in format of DataResponse class.
-
-    See DataMessageFactory for code to instantiate, serialize and deserialize the class
-    """
-
     def __init__(
         self,
         magic: str,
@@ -118,17 +95,8 @@ class DataMessage(CleverMicroMessage):
     def path(self) -> str:
         return self.p_field()
 
-    def routing_key(self) -> str:
-        return sanitize_as_rabbitmq_name(f"{self.d_field()}.{self.p_field()}")
-
 
 class DataResponse(CleverMicroMessage):
-    """
-    The response message to REST & RPC style calls (note: the request is made with DataMessage)
-
-    See DataResponseFactory for code to instantiate, serialize and deserialize the class
-    """
-
     def __init__(
         self,
         magic: str,
@@ -211,6 +179,31 @@ class AMQRoute:
         return self.__str__()
 
 
+@dataclass
+class DataMessage:
+    """
+    The class represents a structure of a message that transfers the user request to the Service.
+    It may also be used to make a call/request between services.
+    The response to this call is in format of DataResponse class.
+    Please see DataMessageFactory for code to instantiate, serialize and deserialize the class
+    """
+
+    magic: str
+    id: SnowflakeId
+    method: str
+    domain: str
+    path: str
+    headers: dict[str, List[str]]
+    trace_info: dict[str, str]
+    base64_body: bytes
+
+    def routing_key(self) -> str:
+        return sanitize_as_rabbitmq_name(f"{self.domain}.{self.path}")
+
+    def body(self) -> bytes:
+        return base64.b64decode(self.base64_body)
+
+
 class MultipartDataMessage(DataMessage):
     def __init__(self, message: DataMessage, extra_data: dict):
         super().__init__(
@@ -226,6 +219,25 @@ class MultipartDataMessage(DataMessage):
         self.extra_data = extra_data
 
 
+@dataclass
+class DataResponse:
+    """
+    The response message to RPC style call (note: the request is made with DataMessage)
+    Please see DataResponseFactory for code to instantiate, serialize and deserialize the class
+    """
+
+    id: str
+    response_code: int
+    content_type: str
+    response: bytes
+    error: str | None
+    error_cause: str | None
+    trace_info: dict[str, str]
+
+    def body(self) -> bytes:
+        return base64.b64decode(self.response)
+
+
 # Ensure backward compatibility
 AMQResponse = DataResponse
 
@@ -293,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/route_database.py

@@ -1,3 +1,4 @@
+import logging
 import re
 from typing import Set
 

amqp/router/router_base.py

@@ -1,7 +1,8 @@
+import logging
 from typing import Callable, List, Set
 
-from amqp.adapter.amq_route_factory import NULL_ROUTE, AMQRouteFactory
-from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
+from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE
+from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.model.model import AMQRoute, DataMessage
 from amqp.router.route_database import RouteDatabase
 from amqp.router.utils import sanitize_as_rabbitmq_name

amqp/router/router_consumer.py

@@ -1,26 +1,26 @@
+import logging
 from typing import Set
 
 from aio_pika.abc import (
-    AbstractIncomingMessage,
     AbstractRobustChannel,
     AbstractRobustQueue,
+    AbstractIncomingMessage,
 )
-
 from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.logging_utils import (
-    logging_debug,
     logging_error,
     logging_info,
+    logging_debug,
     logging_warning,
 )
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQRoute, ServiceMessage
+from amqp.model.model import ServiceMessage, AMQRoute
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import (
-    ANY_RECIPIENT,
     CM_C_REQ_QUEUE,
     CM_REQUEST_EXCHANGE,
     CM_RESPONSE_EXCHANGE,
+    ANY_RECIPIENT,
 )
 from amqp.router.router_producer import RouterProducer
 

amqp/router/router_producer.py

@@ -7,20 +7,21 @@
 * forwards it to the application.
 """
 
+import logging
+
 from aio_pika import Message
 from aio_pika.abc import (
-    AbstractIncomingMessage,
     AbstractRobustChannel,
+    AbstractRobustQueue,
     AbstractRobustConnection,
     AbstractRobustExchange,
-    AbstractRobustQueue,
+    AbstractIncomingMessage, ExchangeType,
 )
-from pika.exchange_type import ExchangeType
 
 from amqp.adapter.logging_utils import (
-    logging_debug,
     logging_error,
     logging_info,
+    logging_debug,
     logging_warning,
 )
 from amqp.adapter.service_message_factory import ServiceMessageFactory
@@ -28,13 +29,13 @@ from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import (
-    ANY_RECIPIENT,
-    CM_C_AMQ_QUEUE,
-    CM_P_REPLY_TO,
-    CM_P_RESP_QUEUE,
+    RouterBase,
     CM_REQUEST_EXCHANGE,
     CM_RESPONSE_EXCHANGE,
-    RouterBase,
+    CM_C_AMQ_QUEUE,
+    CM_P_RESP_QUEUE,
+    CM_P_REPLY_TO,
+    ANY_RECIPIENT,
 )
 
 
@@ -56,7 +57,7 @@ class RouterProducer(RouterBase):
         self.cm_response_exchange: AbstractRobustExchange | None = None
 
         logging_info(
-            "RouterProducer.<init>: %s, route: %s",
+            f"RouterProducer.<init>: %s, route: %s",
             self.amq_configuration.amq_adapter.service_name,
             self.amq_configuration.amq_adapter.route_mapping,
         )
@@ -80,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()
@@ -124,7 +125,7 @@ class RouterProducer(RouterBase):
         * routing data, invokes addRoute or removeRoute depending on the type of the message.
         """
         logging_info(
-            "[%s] Received ROUTING DATA: [%s], msgId=%s",
+            f"[%s] Received ROUTING DATA: [%s], msgId=%s",
             message.delivery_tag,
             message.body,
             message.message_id,
@@ -143,9 +144,9 @@ class RouterProducer(RouterBase):
             cm_message.reply_to,
         )
         if (
-            cm_message.is_valid()
-            and not self.amq_configuration.amq_adapter.service_name
-            == cm_message.reply_to
+                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)
@@ -165,7 +166,7 @@ class RouterProducer(RouterBase):
                 ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
             )
             if not await self.publish_service_message(
-                serviceMessage, self.cm_request_exchange
+                    serviceMessage, self.cm_request_exchange
             ):
                 logging_warning(
                     "RouterProducer could not request route data: %s channel is not open.",
@@ -176,7 +177,7 @@ class RouterProducer(RouterBase):
             logging_error("RouterProducer failed request routing data: %s", ioe)
 
     async def publish_service_message(
-        self, message: ServiceMessage, exchange: AbstractRobustExchange
+            self, message: ServiceMessage, exchange: AbstractRobustExchange
     ) -> bool:
         """
         * Publishes a service message to the service queue and logs success log entry.

amqp/service/amq_message_handler.py

@@ -1,31 +1,32 @@
 import asyncio
-import os
 import threading
 import time
-from asyncio import AbstractEventLoop, Future
+import os
+from asyncio import Future, AbstractEventLoop
 from typing import Dict
 
 from aio_pika import Message
 from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
+
 from opentelemetry import trace
-from opentelemetry.trace import Tracer, TraceState
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+from opentelemetry.trace import Tracer, TraceState
 
 from amqp.adapter.backpressure_handler import BackpressureHandler
-from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.data_response_factory import DataResponseFactory
-from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
-from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
-from amqp.adapter.serializer import map_as_string
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
+from amqp.adapter.file_handler import StreamingFileHandler, FileHandler
+from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.adapter.service_message_factory import ServiceMessageFactory
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import (
-    DataMessage,
     DataResponse,
+    DataMessage,
     MultipartDataMessage,
 )
 from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
+from amqp.adapter.serializer import map_as_string
 
 
 def collect_trace_info():
@@ -74,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.
@@ -265,6 +257,7 @@ class DataMessageHandler:
         :return:
         """
         reply_id = message.correlation_id
+        content_type = message.content_type
         delivery_tag = message.delivery_tag
         debug = f"######[{delivery_tag}] (RPC REPLY) env: {message.headers}, props: {message.properties}, BODY: {message.body.decode()}"
         logging_debug(debug)

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",
+            404,
+            "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",
+            404,
+            "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"
 ]
 

tests/adapter/test_cleverthis_service_adapter.py

@@ -3,7 +3,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 from aiohttp import ClientResponse
 
-from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import DataMessage, DataResponse
 
@@ -174,8 +174,8 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "_authorize_forward_request",
-            wraps=self.adapter._authorize_forward_request,
+            "handle_possible_form",
+            wraps=self.adapter.handle_possible_form,
         ) as mock_handler:
             response = await self.adapter.put(mock_message, None)
 
@@ -199,8 +199,8 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "_authorize_forward_request",
-            wraps=self.adapter._authorize_forward_request,
+            "handle_possible_form",
+            wraps=self.adapter.handle_possible_form,
         ) as mock_handler:
             response = await self.adapter.post(mock_message, None)
 

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