diff --git a/.gitignore b/.gitignore
index 303063d..87af9b9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -55,4 +55,4 @@ hs_err_pid*
 amq_adapter.egg-info/
 build/
 
-
+unused-code/
diff --git a/amqp/adapter/amq_route_factory.py b/amqp/adapter/amq_route_factory.py
index aa1afe2..db6b38c 100644
--- a/amqp/adapter/amq_route_factory.py
+++ b/amqp/adapter/amq_route_factory.py
@@ -2,7 +2,9 @@ import logging
 
 from amqp.model.model import AMQRoute
 
+# record separator
 RS = ":"
+# null route, in lieu of None value
 NULL_ROUTE = {
     "componentName": "",
     "exchange": None,
@@ -14,6 +16,9 @@ NULL_ROUTE = {
 
 
 class AMQRouteFactory:
+    """
+    Define the methods to create and validate AMQRoute objects.
+    """
 
     @staticmethod
     def from_string(data):
diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py
index 7191ecb..7baa410 100644
--- a/amqp/adapter/cleverthis_service_adapter.py
+++ b/amqp/adapter/cleverthis_service_adapter.py
@@ -1,35 +1,33 @@
-import asyncio
+"""
+The layer that interfaces with the CleverThis service that this Adapter is integrating with.
+Methods here will invoke the actual CleverThis Service code, which are passed as Callable object.
+"""
+
 import json
 import logging
-import os
 import re
 import sys
-import threading
 import traceback
-
-from aiohttp import ClientSession, ClientResponse
-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 aiohttp import ClientSession, ClientResponse
+from fastapi import HTTPException
+
+from amqp.adapter import pydantic_serializer
+from amqp.adapter.data_response_factory import DataResponseFactory
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQMessage, AMQResponse
-
-# Track the number of messages currently being processed
-current_parallel_executions = 0
-lock = threading.Lock()
-parallel_workers = os.getenv("PARALLEL_WORKERS", default=10)
+from amqp.model.model import DataMessage, DataResponse
+from amqp.service.multipart_parser import CleverMultiPartParser
 
 
-def parse_request_body(message: AMQMessage):
+def parse_request_body(message: DataMessage):
     """
     Parses the HTTP POST request body based on the MIME type.
 
     Args:
-        message (AMQMessage): Contains raw request and MIME headers.
+        message (DataMessage): Contains raw request and MIME headers.
 
     Returns:
         dict: Parsed data as a dictionary.
@@ -82,53 +80,55 @@ def parse_request_body(message: AMQMessage):
         raise ValueError(f"Unsupported Content-Type: {content_type}")
 
 
-# =============  POST Helper function =============
 async def call_cleverthis_post_put_api(
-        message: AMQMessage,
+        message: DataMessage,
         args: Any,
         api_method: Callable[[Any, Any], Coroutine],
         success_code: int,
         authenticated_user: Any
-) -> AMQResponse:
+) -> DataResponse:
     """
+    Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary.
     Invokes the provided API method with the given arguments and authenticated user,
-    and returns the response as AMQResponse.
-    :param message: AMQMessage containing the details of the call and the payload
+    and returns the response as DataResponse.
+    :param message: DataMessage containing the details of the call and the payload
     :param args: map of the arguments to pass to the API method
     :param api_method: Callable - the method to invoke. Last parameter is the authenticated user
     :param success_code: HTTP code to be returned when operation suceeds
     :param authenticated_user: Authenticated user. The actual class depends on the service
-    :return: AMQResponse class
+    :return: DataResponse class
     """
     try:
         coroutine_call: Coroutine = api_method(args, authenticated_user)
         result: str = await coroutine_call
-        return AMQResponseFactory.of(
+        return DataResponseFactory.of(
             message.id, success_code, message.headers.get("Content-Type", "application/json"),
             result.encode('utf-8'), message.trace_info
         )
     except HTTPException as http_error:
         tb = traceback.extract_tb(sys.exc_info()[2])[-1]
         logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {http_error}")
-        return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain",
+        return DataResponseFactory.of(message.id, http_error.status_code, "text/plain",
                                      str(http_error.detail).encode('utf-8'), message.trace_info)
     except Exception as error:
         tb = traceback.extract_tb(sys.exc_info()[2])[-1]
         logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {error}")
-        return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
+        return DataResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
+
 
 
-# =============  GET Helper function =============
 async def call_cleverthis_get_api(
-        message: AMQMessage,
+        message: DataMessage,
         pattern_or_data: str | dict,
         api_method: Callable[[Tuple[AnyStr | Any, ...], Any], Coroutine],
         authenticated_user: Any
-) -> AMQResponse:
+) -> DataResponse:
     """
-    invokes provided Callable that is expected to be a GET API method, and returns the response as AMQResponse
-    :param message: AMQMessage containing the details of the call
-    :param pattern: If there are path variables, use this pattern to extract them
+    Handles GET requests, which do not contain a body, but may have path variables and extra values
+    in the query string.
+    invokes provided Callable that is expected to be a GET API method, and returns the response as DataResponse
+    :param message: DataMessage containing the details of the call
+    :param pattern_or_data: If there are path variables, use this pattern to extract them
     :param api_method: Callable - the method to invoke. Last parameter is the authenticated user
     :param authenticated_user: Authenticated user. The actual class depends on the service
     :return:
@@ -138,25 +138,26 @@ async def call_cleverthis_get_api(
         if isinstance(pattern_or_data, dict):
             args = pattern_or_data
         else:
-            match = re.search(pattern_or_data, message.path)
+            path, args = pydantic_serializer.parse_url_query(message.path)
+            match = re.search(pattern_or_data, path)
             if match:
-                args = match.groups()
+                args.update(match.groupdict())
             else:
                 _msg = f"Unexpected URL: {message.path}, expected pattern: {pattern_or_data}"
-                return AMQResponseFactory.of(message.id, 404, "text/plain", _msg.encode('utf-8'), message.trace_info)
+                return DataResponseFactory.of(message.id, 404, "text/plain", _msg.encode('utf-8'), message.trace_info)
         crt: Coroutine = api_method(args, authenticated_user)
         result: str = await crt
-        return AMQResponseFactory.of(message.id, 200, "text/plain", result.encode('utf-8'), message.trace_info)
+        return DataResponseFactory.of(message.id, 200, "text/plain", result.encode('utf-8'), message.trace_info)
 
     except HTTPException as http_error:
         tb = traceback.extract_tb(sys.exc_info()[2])[-1]
         logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {http_error}")
-        return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain",
+        return DataResponseFactory.of(message.id, http_error.status_code, "text/plain",
                                      str(http_error.detail).encode('utf-8'), message.trace_info)
     except Exception as error:
         tb = traceback.extract_tb(sys.exc_info()[2])[-1]
         logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {error}")
-        return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
+        return DataResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
 
 
 # =============  CleverThisServiceAdapter =============
@@ -164,58 +165,55 @@ class CleverThisServiceAdapter:
     """
     A base class for CleverThis service adapter that handles the incoming AMQP messages and routes them to the
     appropriate CleverThisService API endpoint.
+    IMPORTANT: This class is not intended to be used directly. It should be subclassed for each CleverThis service
+    and override the methods to handle the specific API calls.
     """
     def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None):
         self.amq_configuration = amq_configuration
         self.session = session
         self.service_host = self.amq_configuration.dispatch.service_host
         self.service_port = self.amq_configuration.dispatch.service_port
+        self.require_authenticated_user = self.amq_configuration.amq_adapter.require_authenticated_user
+
+    def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
+        """
+        Pull the value of the Content-Type header from the message headers.
+        :param message_headers: The headers of the message.
+        return: The content type of the message.
+        """
+        for header, values in message_headers.items():
+            if header.lower() == 'content-type':
+                return values[0].split(';')[0]
+        return 'application/json'
 
     async def get_current_user(self, token: str):
         """
         To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token,
         in format appropriate for the service.
+        :param token: The token to be used for authentication.
         :return: A user object or None if the user is not authenticated.
         """
         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:
+    async def on_message(self, message) -> DataResponse:
         """
         Called when a message is received from the AMQP.
         :param message:
         :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
-            logging.info(f"PARALLEL: Current Capacity [{current_parallel_executions}]")
-            if current_parallel_executions > parallel_workers - 2:
-                logging.warning("Warning: Capacity close to depleted!")
-
-        _auth: str = message.headers.get('Authorization', '')
         _auth_user: Any | None = None
-        if _auth == '' and DEBUG_NO_AUTH == 1:
-            _auth = f"Bearer DEBUG_NO_AUTH"
+        amq_response: DataResponse | None = None
 
-        if 'Bearer' in _auth:
-            try:
-                token = _auth.split(' ')[1]
-                _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:
+        if self.require_authenticated_user:
+            _auth: str = message.headers.get('Authorization', '')
+            if 'Bearer' in _auth:
+                try:
+                    token = _auth.split(' ')[1]
+                    _auth_user = await self.get_current_user(token)
+                except Exception as error:
+                    logging.error(f"Error getting user: {error}")
+
+        if _auth_user or not self.require_authenticated_user:
             method = message.method.lower()
             if method == 'get':
                 amq_response = await self.get(message, _auth_user)
@@ -230,22 +228,21 @@ class CleverThisServiceAdapter:
             else:
                 raise ValueError(f"Unexpected HTTP method: {message.method}")
         else:
-            amq_response = AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
-        with lock:
-            current_parallel_executions -= 1
+            amq_response = DataResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
+
         return amq_response
 
     # ==================================================================================================
     # =====================  C l e v e r T h i s    A P I    m e t h o d s  ============================
     # ==================================================================================================
 
-    async def get(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
+    async def get(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
         Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
         It will attempt to call the REST endpoint directly using HTTP protocol.
         :param message: message from the AMQP that contains the GET request
         :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
+        :return: DataResponse to be sent back to the API Gateway via AMQP
         """
 
         # unmatched path - try to invoke the service directly on this path
@@ -254,19 +251,19 @@ class CleverThisServiceAdapter:
         logging.info(f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}")
         if self.session:
             _http_resp: ClientResponse = await self.session.get(url, headers=headers)
-            return AMQResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type,
+            return DataResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type,
                                          await _http_resp.read(),
                                          message.trace_info)
-        return AMQResponseFactory.of(message.id, 404, "text/plain",
+        return DataResponseFactory.of(message.id, 404, "text/plain",
                                      str("Destination location does not exists").encode('utf-8'), message.trace_info)
 
-    async def put(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
+    async def put(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
         Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
         It will attempt to call the REST endpoint directly using HTTP protocol.
         :param message: message from the AMQP that contains the PUT request
         :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
+        :return: DataResponse to be sent back to the API Gateway via AMQP
         """
         return await self.handle_possible_form(
             message,
@@ -278,13 +275,13 @@ class CleverThisServiceAdapter:
             auth_user
         )
 
-    async def post(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
+    async def post(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
         Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
         It will attempt to call the REST endpoint directly using HTTP protocol.
         :param message: message from the AMQP that contains the POST request, including body and headers
         :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
+        :return: DataResponse to be sent back to the API Gateway via AMQP
         """
         return await self.handle_possible_form(
             message,
@@ -296,32 +293,26 @@ class CleverThisServiceAdapter:
             auth_user
         )
 
-    async def handle_possible_form(self, message: AMQMessage, request_coroutine, auth_user: Any) -> AMQResponse:
+    async def handle_possible_form(self, message: DataMessage, request_coroutine, auth_user: Any) -> DataResponse:
         args = parse_request_body(message)
         return await request_coroutine
 
-    async def head(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
+    async def head(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
         Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
         It will attempt to call the REST endpoint directly using HTTP protocol.
         :param message: message from the AMQP that contains the POST request, including body and headers
         :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
+        :return: DataResponse to be sent back to the API Gateway via AMQP
         """
         raise NotImplementedError
 
-    async def delete(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
+    async def delete(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
         Handles the DELETE requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
         It will attempt to call the REST endpoint directly using HTTP protocol.
         :param message: message from the AMQP that contains the POST request, including body and headers
         :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
+        :return: DataResponse to be sent back to the API Gateway via AMQP
         """
         raise NotImplementedError
-
-    def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
-        for header, values in message_headers.items():
-            if header.lower() == 'content-type':
-                return values[0].split(';')[0]
-        return 'application/json'
diff --git a/amqp/adapter/amq_message_factory.py b/amqp/adapter/data_message_factory.py
similarity index 81%
rename from amqp/adapter/amq_message_factory.py
rename to amqp/adapter/data_message_factory.py
index de943cd..b141a23 100644
--- a/amqp/adapter/amq_message_factory.py
+++ b/amqp/adapter/data_message_factory.py
@@ -1,3 +1,7 @@
+"""
+methods to create, serialize and deserialize DataMessage objects.
+"""
+
 import base64
 import json
 import re
@@ -5,14 +9,14 @@ from io import BytesIO
 from datetime import datetime, timezone
 from typing import Dict, List
 
-from amqp.model.model import AMQMessage
+from amqp.model.model import DataMessage
 from amqp.model.snowflake_id import SnowflakeId
 from amqp.router.serializer import map_with_list_as_string, map_as_string, parse_map_list_string, parse_map_string
 from amqp.router.utils import sanitize_as_rabbitmq_name
 from amqp.adapter.file_downloader import download_buffer
 
 
-class AMQMessageFactory:
+class DataMessageFactory:
     SNOWFLAKE_ID_BITS = 80
     SEQUENCE_BITS = 12
     MACHINE_ID_BITS = 24
@@ -38,7 +42,7 @@ class AMQMessageFactory:
         :return: factory instance
         """
         if cls._instance is None:
-            cls._instance = AMQMessageFactory(machine_id)
+            cls._instance = DataMessageFactory(machine_id)
         return cls._instance
 
     def generate_snowflake_id(self) -> SnowflakeId:
@@ -61,7 +65,7 @@ class AMQMessageFactory:
             self.snowflake_sequence = 0
             self.snowflake_last_timestamp = timestamp
 
-        time = (timestamp - 1000 * int(AMQMessageFactory.snowflake_epoch))
+        time = (timestamp - 1000 * int(DataMessageFactory.snowflake_epoch))
         return SnowflakeId(
             time >> (64 - self.SEQUENCE_BITS - self.MACHINE_ID_BITS),
             ((time << (self.SEQUENCE_BITS + self.MACHINE_ID_BITS)) | self.machine_id | self.snowflake_sequence) & 0xFFFFFFFFFFFFFFFF
@@ -74,7 +78,7 @@ class AMQMessageFactory:
             path: str,
             headers: Dict[str, List[str]],
             message,
-    ) -> AMQMessage:
+    ) -> DataMessage:
         """
         Encapsulate / hide the system level inputs, create the AMQ message from business relevant input
         :param method: a method requested (e.g. HTTP method like GET, POST in case of REST request)
@@ -82,12 +86,12 @@ class AMQMessageFactory:
         :param path: path at the service
         :param headers: HTTP headers
         :param message: payload
-        :return: AMQMessage object
+        :return: DataMessage object
         """
         serializable_headers = headers.copy() if headers else {}
         payload = base64.b64encode(message.encode('utf-8') if isinstance(message, str) else message)
         trace_info = {}
-        return AMQMessage(
+        return DataMessage(
             self.MAGIC_BYTE_HTTP_REQUEST,
             self.generate_snowflake_id(),
             method,
@@ -99,10 +103,10 @@ class AMQMessageFactory:
         )
 
     @staticmethod
-    def amq_message_routing_key(message: AMQMessage) -> str:
+    def amq_message_routing_key(message: DataMessage) -> str:
         """
         Constructs the message specific routing key.
-        :param message: AMQMessage input
+        :param message: DataMessage input
         :return: routing key (as string) compliant to RabbitMQ allowed character set.
         """
         return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
@@ -115,19 +119,19 @@ class AMQMessageFactory:
         :return: the message timestamp
         """
         bits = _id.get_bits()
-        low = bits[0] >> (AMQMessageFactory.SEQUENCE_BITS + AMQMessageFactory.MACHINE_ID_BITS)
-        high = bits[1] << (64 - AMQMessageFactory.SEQUENCE_BITS - AMQMessageFactory.MACHINE_ID_BITS)
-        return (high | low) + AMQMessageFactory.snowflake_epoch * 1000
+        low = bits[0] >> (DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS)
+        high = bits[1] << (64 - DataMessageFactory.SEQUENCE_BITS - DataMessageFactory.MACHINE_ID_BITS)
+        return (high | low) + DataMessageFactory.snowflake_epoch * 1000
 
     @staticmethod
-    def to_string(message: AMQMessage) -> str:
+    def to_string(message: DataMessage) -> str:
         """
         Human-readable representation.
-        :param message: AMQMessage input
+        :param message: DataMessage input
         :return: as string
         """
         return (
-            f"AMQMessage{{"
+            f"DataMessage{{"
             f"id={message.id}, "
             f"method='{message.method}', "
             f"domain='{message.domain}', "
@@ -139,9 +143,9 @@ class AMQMessageFactory:
         )
 
     @staticmethod
-    def serialize(message: AMQMessage) -> bytes:
+    def serialize(message: DataMessage) -> bytes:
         """
-        Converts AMQMessage to byte array
+        Converts DataMessage to byte array
         :param message: message to serialize
         :return: messages as bytes
         """
@@ -154,24 +158,24 @@ class AMQMessageFactory:
         msg_length = 2 + header_length + len(message.base64_body)
         with BytesIO(bytearray(msg_length)) as baos:
             baos.write(prolog_len)
-            baos.write(AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
+            baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
             baos.write(id_bytes)
             baos.write(prolog)
             baos.write(message.base64_body)
             return baos.getvalue()
 
     @staticmethod
-    def from_bytes(input_bytes: bytes) -> AMQMessage:
+    def from_bytes(input_bytes: bytes) -> DataMessage:
         """
-        Builds the AMQMessage from its serialized (byte array) form.
+        Builds the DataMessage from its serialized (byte array) form.
         :param input_bytes: serialized message
-        :return: AMQMessage instance
+        :return: DataMessage instance
         """
         prolog_len = (input_bytes[0] << 8) + input_bytes[1]
         metadata = input_bytes[2:prolog_len + 2].decode()
 
         pattern = re.compile(
-            f"{AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST}"
+            f"{DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST}"
             r"([0-9A-Fa-f]+)\|"
             r"m=([^|]*)\|"
             r"d=([^|]*)\|"
@@ -183,7 +187,7 @@ class AMQMessageFactory:
         match = pattern.match(metadata)
         if not match:
             raise ValueError(
-                f"AMQMessage format does not match expected format. "
+                f"DataMessage format does not match expected format. "
                 f"Input string [{metadata}] does not match pattern: {pattern.pattern}"
             )
 
@@ -198,8 +202,8 @@ class AMQMessageFactory:
         headers = parse_map_list_string(headers_str)
         trace_info = parse_map_string(trace_info_str)
 
-        return AMQMessage(
-            AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST,
+        return DataMessage(
+            DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST,
             SnowflakeId.from_hex(id_str),
             method,
             domain,
@@ -210,12 +214,12 @@ class AMQMessageFactory:
         )
 
     @staticmethod
-    async def from_stream(stream: bytes, rabbitmq_url: str, loop) -> AMQMessage | None:
+    async def from_stream(stream: bytes, rabbitmq_url: str, loop) -> DataMessage | None:
         """
-        Builds the AMQMessage from its serialized (byte array) form.
+        Builds the DataMessage from its serialized (byte array) form.
         :param stream: serialized message
         :param rabbitmq_url: RabbitMQ connection URL
-        :return: AMQMessage instance
+        :return: DataMessage instance
         """
         # async implementation
         message_data = json.loads(stream.decode())
@@ -223,5 +227,5 @@ class AMQMessageFactory:
         if req_info:
             (body, eof) = await download_buffer(loop=loop, rabbitmq_url=rabbitmq_url, queue_name=req_info)
             if eof == 1:
-                return AMQMessageFactory.from_bytes(body)
+                return DataMessageFactory.from_bytes(body)
         return None
diff --git a/amqp/adapter/amq_response_factory.py b/amqp/adapter/data_response_factory.py
similarity index 75%
rename from amqp/adapter/amq_response_factory.py
rename to amqp/adapter/data_response_factory.py
index 59391cb..60483dd 100644
--- a/amqp/adapter/amq_response_factory.py
+++ b/amqp/adapter/data_response_factory.py
@@ -1,37 +1,41 @@
+"""
+method to create, serialize and deserialize DataResponse message
+"""
+
 import base64
 import re
 from io import BytesIO
-from amqp.model.model import AMQResponse
+from amqp.model.model import DataResponse
 from amqp.model.snowflake_id import SnowflakeId
 from amqp.router.serializer import parse_map_string, map_as_string
 
 
-class AMQResponseFactory:
+class DataResponseFactory:
     MAGIC_BYTE_HTTP_REQUEST = "A"
 
     @staticmethod
-    def create_async_response_message(_id: SnowflakeId) -> AMQResponse:
+    def create_async_response_message(_id: SnowflakeId) -> DataResponse:
         """
         Helper method to create standard OK response.
         :param _id: ID
         :return: Response message
         """
-        return AMQResponse(
+        return DataResponse(
             str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {}
         )
 
     @staticmethod
-    def of(id: SnowflakeId, response_code: int, content_type: str, body: bytes, trace_info: dict[str,str]) -> AMQResponse:
+    def of(id: SnowflakeId, response_code: int, content_type: str, body: bytes, trace_info: dict[str,str]) -> DataResponse:
         """
-        Create the AMQResponse message supplying all key values
+        Create the DataResponse message supplying all key values
         :param id: SnowflakeID
         :param response_code: HTTP response code
         :param content_type: MIME content type
         :param body: payload
         :param trace_info: Jaeger trace info (id)
-        :return: AMQResponseMessage object
+        :return: DataResponseMessage object
         """
-        return AMQResponse(
+        return DataResponse(
             id=id.as_string(),
             response_code=response_code,
             content_type=content_type,
@@ -42,7 +46,7 @@ class AMQResponseFactory:
         )
 
     @staticmethod
-    def serialize(response: AMQResponse) -> bytes:
+    def serialize(response: DataResponse) -> bytes:
         """
         Serialize the message to byte array (to be sent over RabbitMQ)
         :param response: Object to serialize
@@ -57,24 +61,24 @@ class AMQResponseFactory:
         msg_length = 2 + header_length + len(response.response)
         with BytesIO(bytearray(msg_length)) as baos:
             baos.write(prolog_len)
-            baos.write(AMQResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
+            baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
             baos.write(id_bytes)
             baos.write(prolog)
             baos.write(base64.b64encode(response.response))
             return baos.getvalue()
 
     @staticmethod
-    def from_bytes(input_bytes: bytes) -> AMQResponse:
+    def from_bytes(input_bytes: bytes) -> DataResponse:
         """
-        Buil;ds the AMQResponse object from its serialized form
+        Buil;ds the DataResponse object from its serialized form
         :param input_bytes: byte array - serialized response
-        :return: AMQResponse object
+        :return: DataResponse object
         """
         prolog_len = (input_bytes[0] << 8) + input_bytes[1]
         metadata = input_bytes[2:prolog_len + 2].decode()
 
         pattern = re.compile(
-            f"{AMQResponseFactory.MAGIC_BYTE_HTTP_REQUEST}"
+            f"{DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST}"
             r"([0-9A-Fa-f]+)\|"
             r"r=([^|]*)\|"
             r"c=([^|]*)\|"
@@ -86,7 +90,7 @@ class AMQResponseFactory:
         match = pattern.match(metadata)
         if not match:
             raise ValueError(
-                f"AMQResponse format does not match expected format. "
+                f"DataResponse format does not match expected format. "
                 f"Input string [{metadata}] does not match pattern: {pattern.pattern}"
             )
 
@@ -100,7 +104,7 @@ class AMQResponseFactory:
 
         trace_info = parse_map_string(trace_info_str)
 
-        return AMQResponse(
+        return DataResponse(
             id_str,
             response_code,
             content_type,
@@ -109,3 +113,7 @@ class AMQResponseFactory:
             error_cause,
             trace_info,
         )
+
+
+# Ensure backward compatibility
+AMQResponseFactory = DataResponseFactory
diff --git a/amqp/adapter/file_downloader.py b/amqp/adapter/file_downloader.py
index ff078c6..f37f66a 100644
--- a/amqp/adapter/file_downloader.py
+++ b/amqp/adapter/file_downloader.py
@@ -1,3 +1,7 @@
+"""
+implements the file downloader for the amqp adapter, where file is sent in chunks over RabbitMQ
+"""
+
 import asyncio
 import json
 import os
diff --git a/amqp/adapter/multipart_download.py b/amqp/adapter/multipart_download.py
deleted file mode 100644
index 6f64d4c..0000000
--- a/amqp/adapter/multipart_download.py
+++ /dev/null
@@ -1,140 +0,0 @@
-import asyncio
-import json
-import os
-import threading
-from collections import defaultdict
-
-from aio_pika.abc import HeadersType, AbstractIncomingMessage, AbstractQueue, AbstractRobustChannel
-from aiormq import Channel
-
-
-class MultipartDownload:
-
-    def __init__(self, headers: HeadersType, channel: AbstractRobustChannel):
-        self.headers = headers
-        self.channel = channel
-        self.addressing: int = headers.get('clevermicro.addressing', 0)
-        self.files_to_download = []
-        self.threads = []
-        # Global dictionary to track completed downloads
-        self.download_status = defaultdict(lambda: defaultdict(bool))
-        self.future = None
-
-    def ensure_directory_exists(self, filepath):
-        """Ensures that the directory for the given filepath exists."""
-        dir_name = os.path.dirname(filepath)
-        if dir_name and not os.path.exists(dir_name):
-            os.makedirs(dir_name)
-
-    async def start_queue2(self, channel: Channel, queue_name: str, callback):
-        dok = await channel.queue_declare(queue=queue_name, durable=False, auto_delete=True)
-        await channel.basic_consume(queue=queue_name, consumer_callback=callback, no_ack=True, exclusive=True)
-
-    async def start_queue(self, queue_name: str, callback):
-        """
-        Starts consuming from a queue.
-
-        Args:
-            :param queue_name: The name of the queue to consume from.
-            :param callback: The callback function to call when a message is received.
-        """
-        queue: AbstractQueue = await self.channel.declare_queue(name=queue_name, durable=False, auto_delete=True)
-        await queue.consume(callback=callback, no_ack=True, exclusive=True)
-
-    def download_file(self, channel: Channel, file_info, message_data, output_dir):
-        """
-        Downloads a file from RabbitMQ based on the provided file information.
-
-        Args:
-            channel: The RabbitMQ channel.
-            file_info (dict): A dictionary containing file metadata
-            message_data (dict): the entire message
-            output_dir (str): The directory where the file should be saved.
-        """
-        filename = file_info['filename']
-        size = file_info['size']
-        stream_name = file_info['streamName']
-        filepath = os.path.join(output_dir, filename)
-        self.ensure_directory_exists(filepath)
-
-        print(f"Downloading file: {filename} (size: {size} bytes) from stream: {stream_name} to {filepath}")
-
-        bytes_received = 0
-        with open(filepath, 'wb') as f:
-            async def callback(in_message: AbstractIncomingMessage):
-                nonlocal bytes_received
-                f.write(in_message.body)
-                bytes_received += len(in_message.body)
-                # print(f"Received {len(body)} bytes for {filename}. Total: {bytes_received}/{size}") # uncomment for verbose
-
-                if bytes_received >= size:
-                    print(f"Finished downloading {filename}")
-                    f.close()
-                    # Update global download status
-                    message_req_info = json.loads(message_data.get('req-info', '{}'))
-                    if message_req_info:
-                        self.download_status[message_req_info['from']][filename] = True
-                    await in_message.channel.close()  # stop consuming after the file is downloaded
-
-            self.start_queue2(channel, stream_name, callback)
-
-    def on_message(self, in_message: AbstractIncomingMessage, body: bytes) -> asyncio.Future:
-        """
-        Callback function for handling messages from the 'amq-cleverswarm' queue.
-
-        Args:
-            ch: The RabbitMQ channel.
-            method: The delivery method.
-            properties: The message properties.
-            body: The message body (a JSON string).
-            :param body:
-        """
-        message_data = json.loads(body.decode())
-        req_info = json.loads(message_data.get('req-info', '{}')) # Handle missing 'req-info'
-        self.future = asyncio.Future()
-
-        if not req_info:
-            print("Received message without 'req-info', skipping.")
-            in_message.ack(False)
-            self.future.set_result(False)
-            return self.future
-
-        for file_key in req_info['files']:
-            files_info = req_info['files'].get(file_key, [])  # Handle missing file types
-            self.files_to_download.extend(files_info)
-
-        if not self.files_to_download:
-            print("No files to download in this message.")
-            self.future.set_result(False)
-            return self.future
-
-        # Create a thread for each file download
-        output_dir = "downloaded_files"  # You can change this
-        for file_info in self.files_to_download:
-            thread = threading.Thread(
-                target=self.download_file,
-                args=(in_message.channel, file_info, message_data, output_dir)
-            )
-            self.threads.append(thread)
-            thread.start()
-
-        # Wait for all download threads to complete
-        asyncio.run(self.wait_for_download_completion(self.threads))
-
-        # After all files are downloaded, acknowledge the original message
-        print("All files are being downloaded. Acknowledging the main message.")
-        in_message.ack(False)
-        print(f"Message from {in_message.routing_key} acknowledged.")
-        return self.future
-
-    async def wait_for_download_completion(self, threads):
-        """
-        Waits for all download threads to complete.
-
-        Args:
-            threads (list): A list of threads to wait for.
-        """
-        for thread in threads:
-            thread.join()
-
-        self.future.set_result(True)
diff --git a/amqp/adapter/pydantic_serializer.py b/amqp/adapter/pydantic_serializer.py
index 3734926..0c19967 100644
--- a/amqp/adapter/pydantic_serializer.py
+++ b/amqp/adapter/pydantic_serializer.py
@@ -1,3 +1,7 @@
+"""
+This module provides serialization and deserialization functions for Pydantic models that extend BaseModel
+"""
+
 import uuid
 import json
 from datetime import datetime
diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py
index 2e8e3d8..ef78102 100644
--- a/amqp/adapter/service_message_factory.py
+++ b/amqp/adapter/service_message_factory.py
@@ -1,9 +1,9 @@
 import base64
 import logging
 
-from amqp.adapter.amq_message_factory import AMQMessageFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.trace_info_adapter import TraceInfoAdapter
-from amqp.model.model import CleverMicroServiceMessage
+from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
 from amqp.router.serializer import parse_map_string, map_as_string
@@ -12,40 +12,40 @@ RS = "|"
 SPLIT_RS = r"\|"
 
 
-INVALID_MESSAGE = CleverMicroServiceMessage()  # default values mean invalid message
+INVALID_MESSAGE = ServiceMessage()  # default values mean invalid message
 
 
-class CleverMicroServiceMessageFactory:
+class ServiceMessageFactory:
     """
     build/serialize/deserialize CleverMicro Service Message class
     """
     def __init__(self, name: str):
         self.process_name = name
 
-    def to_string(self, message: CleverMicroServiceMessage) -> str:
+    def to_string(self, message: ServiceMessage) -> str:
         """
-        Convert CleverMicroServiceMessage to human-readable format
-        :param message: CleverMicroServiceMessage object
+        Convert ServiceMessage to human-readable format
+        :param message: ServiceMessage object
         :return: as string
         """
         if message is not None:
             return (
-                f"CleverMicroServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|"
+                f"ServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|"
                 f"recipient={message.recipient_name}|replyTo={message.reply_to}|"
                 f"traceInfo={message.trace_info}|body={base64.b64encode(message.message.encode()).decode() if message.payload_type & 0x80 else message.message}}}"
             )
         return "null"
 
-    def of(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage:
+    def of(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> ServiceMessage:
         """
-        Builds CleverMicroServiceMessage object from relevant values.
+        Builds ServiceMessage object from relevant values.
         :param message_type: enum Message type
         :param payload: Message content, stored as-is
         :param recipient_name: Recipient name (from the routing expression)
-        :return: CleverMicroServiceMessage  object
+        :return: ServiceMessage  object
         """
-        _id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
-        return CleverMicroServiceMessage(
+        _id = DataMessageFactory.get_instance(1).generate_snowflake_id()
+        return ServiceMessage(
             id=_id,
             payload_type=0,
             message_type=message_type,
@@ -55,16 +55,16 @@ class CleverMicroServiceMessageFactory:
             message=payload
         )
 
-    def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage:
+    def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> ServiceMessage:
         """
-        Builds CleverMicroServiceMessage object from relevant values, stores content Base64 encoded.
+        Builds ServiceMessage object from relevant values, stores content Base64 encoded.
         :param message_type: enum Message type
         :param payload: Message content, stored as Base64 string
         :param recipient_name: Recipient name (from the routing expression)
-        :return: CleverMicroServiceMessage  object
+        :return: ServiceMessage  object
         """
-        _id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
-        return CleverMicroServiceMessage(
+        _id = DataMessageFactory.get_instance(1).generate_snowflake_id()
+        return ServiceMessage(
             id=_id,
             payload_type=0x80,
             message_type=message_type,
@@ -74,13 +74,13 @@ class CleverMicroServiceMessageFactory:
             message=payload
         )
 
-    def from_bytes(self, input_bytes: bytes) -> CleverMicroServiceMessage:
+    def from_bytes(self, input_bytes: bytes) -> ServiceMessage:
         """
-        Deserialize CleverMicroServiceMessage from byte array serialized form.
-        :param input_bytes: serialized CleverMicroServiceMessage
-        :return: CleverMicroServiceMessage object represented by the input.
+        Deserialize ServiceMessage from byte array serialized form.
+        :param input_bytes: serialized ServiceMessage
+        :return: ServiceMessage object represented by the input.
         """
-        logging.info(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]")
+        logging.info(f"ServiceMessage >>> [{input_bytes.decode()}]")
         input_str_array = input_bytes.decode().split(RS)
         try:
             i = 0
@@ -96,7 +96,7 @@ class CleverMicroServiceMessageFactory:
             trace_info = parse_map_string(input_str_array[i]) if i < len(input_str_array) else {}
             i += 1
             message = input_str_array[i] if i < len(input_str_array) else ""
-            return CleverMicroServiceMessage(
+            return ServiceMessage(
                 id=SnowflakeId.from_hex(_id),
                 payload_type=payload_type,
                 message_type=message_type,
@@ -114,10 +114,10 @@ class CleverMicroServiceMessageFactory:
             )
             return INVALID_MESSAGE
 
-    def serialize(self, message: CleverMicroServiceMessage) -> bytes:
+    def serialize(self, message: ServiceMessage) -> bytes:
         """
-        Serialize CleverMicroServiceMessage to byte array.
-        :param message: CleverMicroServiceMessage to serialize.
+        Serialize ServiceMessage to byte array.
+        :param message: ServiceMessage to serialize.
         :return: serialized object.
         """
         return (
@@ -126,8 +126,7 @@ class CleverMicroServiceMessageFactory:
             f"{base64.b64encode(message.message.encode()).decode() if message.payload_type != 0 else message.message}"
         ).encode('utf-8')
 
-
-    def format_message_type(self, message: CleverMicroServiceMessage) -> str:
+    def format_message_type(self, message: ServiceMessage) -> str:
         """
         Formats the Enum message type and decorates it with base64 flag at highest bit.
         For compatibility with 0-based enum ordinals in Java subtract 1 from ordinal value.
diff --git a/amqp/adapter/test_amq_message_factory.py b/amqp/adapter/test_data_message_factory.py
similarity index 87%
rename from amqp/adapter/test_amq_message_factory.py
rename to amqp/adapter/test_data_message_factory.py
index 898696d..d430bf3 100644
--- a/amqp/adapter/test_amq_message_factory.py
+++ b/amqp/adapter/test_data_message_factory.py
@@ -1,25 +1,25 @@
 from datetime import datetime, timezone
 from unittest import TestCase
 
-from amqp.adapter.amq_message_factory import AMQMessageFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
 
 
-class AMQMessageFactoryTest(TestCase):
+class DataMessageFactoryTest(TestCase):
     def test_from_bytes(self):
         raw = "\002\027A64f3b5af4d00000a4000|m=POST|d=clever-pybook.localhost|p=/debug|h={Accept=[*/*],X-Forwarded-Host=[clever-pybook.localhost:8085],User-Agent=[curl/8.7.1],X-Forwarded-Proto=[http],X-Forwarded-For=[172.18.0.1],Host=[clever-pybook.localhost:8085],Accept-Encoding=[gzip],Content-Length=[1475],X-Forwarded-Port=[8085],X-Real-Ip=[172.18.0.1],X-Forwarded-Server=[c5945bef67d1],Content-Type=[multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v]}|t={traceparent=00-bed6eb7de0d8ccf4c80b145a4bfec357-8d1f3bafc471bf0f-01}|b=TEST_ME"
-        msg = AMQMessageFactory.from_bytes(raw.encode("utf-8"))
+        msg = DataMessageFactory.from_bytes(raw.encode("utf-8"))
         self.assertIsNotNone(msg.id)
         self.assertEqual(msg.id.as_string(),'64F3B5AF4D00000A4000'.lower(),'SnowflakeID incorrectly parsed.')
         self.assertEqual(msg.headers["Content-Type"][0], "multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v")
 
     def test_generate_snowflake_id(self):
-        _id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
+        _id = DataMessageFactory.get_instance(1).generate_snowflake_id()
         self.assertEqual('00001000', _id.as_string()[-8:], 'Instance ID not set correctly')
 
-        _id = AMQMessageFactory(2).generate_snowflake_id()
+        _id = DataMessageFactory(2).generate_snowflake_id()
         self.assertEqual('00002000', _id.as_string()[-8:], 'Instance ID not set correctly')
 
-        _f = AMQMessageFactory(127)
+        _f = DataMessageFactory(127)
         _i0 = _f.generate_snowflake_id()
         _i1 = _f.generate_snowflake_id()
         _i2 = _f.generate_snowflake_id()
@@ -30,7 +30,7 @@ class AMQMessageFactoryTest(TestCase):
         self.assertEqual('0007F003'.lower(), _i3.as_string()[-8:], 'Instance ID and sequence not set correctly')
 
     def test_create_request_message(self):
-        _f = AMQMessageFactory(34)
+        _f = DataMessageFactory(34)
         _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
         self.assertEqual("localhost", _req.domain, 'Domain not set correctly')
         self.assertEqual("/test/me", _req.path, 'Context path not set correctly')
@@ -38,7 +38,7 @@ class AMQMessageFactoryTest(TestCase):
         self.assertEqual(b'Quick brown fox', _req.body(), "Body not correctly decoded")
 
     def test_serialize(self):
-        _f = AMQMessageFactory(34)
+        _f = DataMessageFactory(34)
         _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
         _ser = _f.serialize(_req)
         self.assertEqual(65, _ser[2])
@@ -50,19 +50,19 @@ class AMQMessageFactoryTest(TestCase):
         self.assertEqual("v2", _deser.headers["k2"][0])
 
     def test_amq_message_routing_key(self):
-        _f = AMQMessageFactory(35)
+        _f = DataMessageFactory(35)
         _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
         _key = _f.amq_message_routing_key(_req)
         self.assertEqual("localhost.test.me", _key, "Message routing key wrongly extracted from the message")
 
     def test_get_timestamp_from_id(self):
-        _f = AMQMessageFactory(35)
+        _f = DataMessageFactory(35)
         _timestamp = _f.get_timestamp_from_id(_f.generate_snowflake_id())
         _x_timestamp = int(datetime.now(timezone.utc).timestamp() * 1000)
         self.assertLessEqual(_x_timestamp - _timestamp, 1, "Timestamp in SnowflakeId differs too much from current time")
 
     def test_to_string(self):
-        _f = AMQMessageFactory(35)
+        _f = DataMessageFactory(35)
         _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
         _as_str = _f.to_string(_req)
         self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found")
diff --git a/amqp/adapter/test_amq_response_factory.py b/amqp/adapter/test_data_response_factory.py
similarity index 71%
rename from amqp/adapter/test_amq_response_factory.py
rename to amqp/adapter/test_data_response_factory.py
index ec8b515..a6e5c83 100644
--- a/amqp/adapter/test_amq_response_factory.py
+++ b/amqp/adapter/test_data_response_factory.py
@@ -1,22 +1,22 @@
 from unittest import TestCase
 
-from amqp.adapter.amq_message_factory import AMQMessageFactory
-from amqp.adapter.amq_response_factory import AMQResponseFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
+from amqp.adapter.data_response_factory import DataResponseFactory
 from amqp.model.snowflake_id import SnowflakeId
 
 
-class TestAMQResponseFactory(TestCase):
+class TestDataResponseFactory(TestCase):
 
     def test_create_async_response_message(self):
-        _f = AMQResponseFactory()
-        _g = AMQMessageFactory(111)
+        _f = DataResponseFactory()
+        _g = DataMessageFactory(111)
         _resp = _f.create_async_response_message(_g.generate_snowflake_id())
         self.assertEqual(200, _resp.response_code)
         self.assertEqual(b"OK", _resp.response)
 
     def test_serialize(self):
-        _f = AMQResponseFactory()
-        _g = AMQMessageFactory(111)
+        _f = DataResponseFactory()
+        _g = DataMessageFactory(111)
         _snowflakeId: SnowflakeId = _g.generate_snowflake_id()
         _resp = _f.create_async_response_message(_snowflakeId)
         _ser = _f.serialize(_resp)
@@ -25,12 +25,12 @@ class TestAMQResponseFactory(TestCase):
         self.assertEqual((_ser[0]+_ser[1]), len(_ser) - 2 - len('T0s='))
 
     def test_from_bytes(self):
-        _f = AMQResponseFactory()
-        _g = AMQMessageFactory(111)
+        _f = DataResponseFactory()
+        _g = DataMessageFactory(111)
         _snowflakeId: SnowflakeId = _g.generate_snowflake_id()
         _resp = _f.create_async_response_message(_snowflakeId)
         _ser = _f.serialize(_resp)
         _deser = _f.from_bytes(_ser)
         self.assertEqual(_snowflakeId.as_string(), _deser.id)
         self.assertEqual(200, _deser.response_code)
-        self.assertEqual(b'OK', _deser.response)
\ No newline at end of file
+        self.assertEqual(b'OK', _deser.response)
diff --git a/amqp/adapter/test_pydantic_serializer.py b/amqp/adapter/test_pydantic_serializer.py
new file mode 100644
index 0000000..2e56136
--- /dev/null
+++ b/amqp/adapter/test_pydantic_serializer.py
@@ -0,0 +1,36 @@
+from unittest import TestCase
+
+import pytest
+
+from amqp.adapter.pydantic_serializer import parse_url_query
+
+
+class PydanticSerializerTest(TestCase):
+    def test_serialize_object(self):
+        assert False
+
+    def test_deserialize_object(self):
+        assert False
+
+    def test__convert_value(self):
+        assert False
+
+    def test__is_uuid_string(self):
+        assert False
+
+    def test__is_datetime_string(self):
+        assert False
+
+    def test__is_int_string(self):
+        assert False
+
+    def test_serialize_to_json(self):
+        assert False
+
+    def test_parse_url_query(self):
+        path, args = parse_url_query("http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741?response_type=JsonFile")
+        assert path == "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741"
+        assert args == {'response_type': 'JsonFile'}
+        path, args = parse_url_query("http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741")
+        assert path == "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741"
+        assert args == {}
diff --git a/amqp/adapter/test_service_message_factory.py b/amqp/adapter/test_service_message_factory.py
index c049622..3dfaa8e 100644
--- a/amqp/adapter/test_service_message_factory.py
+++ b/amqp/adapter/test_service_message_factory.py
@@ -1,19 +1,19 @@
 from unittest import TestCase
 
-from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory, RS, INVALID_MESSAGE
-from amqp.model.model import CleverMicroServiceMessage
+from amqp.adapter.service_message_factory import ServiceMessageFactory, RS, INVALID_MESSAGE
+from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 
 
-class CleverMicroServiceMessageTest(TestCase):
+class ServiceMessageTest(TestCase):
 
     def test_to_string(self):
-        _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory('bumble-bee')
-        message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
+        _f: ServiceMessageFactory = ServiceMessageFactory('bumble-bee')
+        message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
         message_str = _f.to_string(message)
         self.assertTrue("|type=01|" in message_str)
         self.assertTrue("|body=Duck" in message_str)
-        message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald")
+        message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald")
         message_str = _f.to_string(message)
         self.assertTrue("|type=03|" in message_str)
         self.assertTrue("|body=Duck" in message_str)
@@ -22,7 +22,7 @@ class CleverMicroServiceMessageTest(TestCase):
         self.assertEqual("null", nullMsg)
 
     def test_from(self):
-        _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
+        _f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
         message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
         serialized = _f.serialize(message)
         deserialized = _f.from_bytes(serialized)
@@ -35,7 +35,7 @@ class CleverMicroServiceMessageTest(TestCase):
         self.assertEqual(invalidMsg, INVALID_MESSAGE)
 
     def test_base64(self):
-        _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
+        _f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
         message = _f.as_base64(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
         serialized = _f.serialize(message)
         assert b"|81|" in serialized
@@ -45,7 +45,7 @@ class CleverMicroServiceMessageTest(TestCase):
         assert deserialized.message == "Duck"
     
     def test_serialize(self):
-        _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
+        _f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
         message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
         serialized = _f.serialize(message)
         assert RS + "0" + str(ServiceMessageType.ROUTING_DATA.value)
diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py
index 69cf236..9772298 100644
--- a/amqp/config/amq_configuration.py
+++ b/amqp/config/amq_configuration.py
@@ -25,6 +25,9 @@ class AMQAdapter:
         self.service_name = config.get('CleverMicro-AMQ', self.PREFIX + 'service-name',
                                        fallback='amq-python-adapter-' + str(self.generator_id))
         self.route_mapping = config.get('CleverMicro-AMQ', self.PREFIX + 'route-mapping', fallback=None)
+        self.require_authenticated_user = config.get(
+            'CleverMicro-AMQ', self.PREFIX + 'require-authenticated-user', fallback=False
+        )
 
     def adapter_prefix(self) -> str:
         return f"{self.service_name}-{self.generator_id}-"
@@ -72,13 +75,12 @@ class Backpressure:
         """
         cm.backpressure.threshold=20
         cm.backpressure.time-window=3000
-        cm.backpressure.management-service-url=management-service:8080
 
         :param config: config parser to read configuration values
         """
-        self.management_service_url = config.get('CleverMicro-AMQ', self.PREFIX + 'management-service-url', fallback='management-service:8080')
-        self.threshold = config.getint('CleverMicro-AMQ', self.PREFIX + 'threshold', fallback=20)
-        self.time_window = config.getint('CleverMicro-AMQ', self.PREFIX + 'time-window', fallback=3000)
+        self.threshold = config.getint('CleverMicro-AMQ', self.PREFIX + 'threshold', fallback=0)
+        # time-window is in milliseconds, to report backpressure IDLE alert
+        self.time_window = config.getint('CleverMicro-AMQ', self.PREFIX + 'time-window', fallback=10000)
 
 
 class AMQConfiguration:
diff --git a/amqp/config/application.properties b/amqp/config/application.properties
index 604e982..32b8f79 100644
--- a/amqp/config/application.properties
+++ b/amqp/config/application.properties
@@ -3,16 +3,20 @@
 # ====================================================================
 [CleverMicro-AMQ]
 cm.amq-adapter.service-name=amq-adapter-python
-cm.amq-adapter.is-router=false
 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
 
 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 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=cleverswarm.localhost
+#cm.dispatch.service-port=8080
+#
+# Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert
 cm.backpressure.threshold=20
-cm.backpressure.time-window=3000
-cm.backpressure.management-service-url=management-service:8080
diff --git a/amqp/config/application.properties.local b/amqp/config/application.properties.local
index 4f603b0..a4ff6aa 100644
--- a/amqp/config/application.properties.local
+++ b/amqp/config/application.properties.local
@@ -3,7 +3,6 @@
 # ====================================================================
 [CleverMicro-AMQ]
 cm.amq-adapter.service-name=amq-adapter-python
-cm.amq-adapter.is-router=false
 cm.amq-adapter.generator-id=211
 cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:localhost.#:5:0
 
@@ -12,7 +11,9 @@ cm.dispatch.use-confirms=false
 cm.dispatch.amq-host=localhost
 cm.dispatch.amq-port=5672
 cm.dispatch.amq-port-tls=5671
+# 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=cleverswarm.localhost
+#cm.dispatch.service-port=8080
 
 cm.backpressure.threshold=20
-cm.backpressure.time-window=3000
-cm.backpressure.management-service-url=management-service:8080
diff --git a/amqp/model/model.py b/amqp/model/model.py
index cebb92c..5743c58 100644
--- a/amqp/model/model.py
+++ b/amqp/model/model.py
@@ -1,4 +1,8 @@
 import base64
+import json
+from enum import Enum
+from dataclasses import dataclass
+from typing import Final
 
 from typing import List, Dict
 from dataclasses import dataclass, field
@@ -139,12 +143,12 @@ class AMQRoute:
 
 
 @dataclass
-class AMQMessage:
+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 AMQResponse class.
-    Please see AMQMessageFactory for code to instantiate, serialize and deserialize the class
+    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
@@ -162,19 +166,24 @@ class AMQMessage:
         return base64.b64decode(self.base64_body)
 
 
-class MultipartAMQMessage(AMQMessage):
-    def __init__(self, message: AMQMessage, extra_data: dict):
+# Ensure backward compatibility
+AMQMessage = DataMessage
+
+
+class MultipartDataMessage(DataMessage):
+    def __init__(self, message: DataMessage, extra_data: dict):
         super().__init__(message.magic, message.id,
                          message.method, message.domain, message.path,
                          message.headers, message.trace_info,
                          message.base64_body)
         self.extra_data = extra_data
 
+
 @dataclass
-class AMQResponse:
+class DataResponse:
     """
-    The response message to RPC style call (note: the request is made with AMQMessage)
-    Please see AMQResponseFactory for code to instantiate, serialize and deserialize the class
+    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
@@ -188,11 +197,15 @@ class AMQResponse:
         return base64.b64decode(self.response)
 
 
+# Ensure backward compatibility
+AMQResponse = DataResponse
+
+
 @dataclass(frozen=True)
-class CleverMicroServiceMessage:
+class ServiceMessage:
     """
     The CleverMicro service message that is used to communicate the internal state of the AMQ channel.
-    In v0.1 it supports service discovery / route registration and Backpressure status notification.
+    In v0.2 it supports service discovery / route registration and Backpressure status notification.
     """
     id: SnowflakeId = None
     payload_type: int = 0
@@ -205,3 +218,41 @@ class CleverMicroServiceMessage:
     def is_valid(self) -> bool:
         return self.message_type is not None
 
+
+class ScalingRequestAlert(Enum):
+    OVERLOAD = "OVERLOAD"
+    IDLE = "IDLE"
+    UPDATE = "UPDATE"
+
+
+@dataclass(frozen=True)
+class ScalingRequest:
+    service_id: str
+    task_id: str
+    max_availability: int
+    current_availability: int
+    request_type: ScalingRequestAlert
+    version: int = 1
+
+    def to_json(self) -> str:
+        """Converts the object to a JSON string matching the Java structure"""
+        return json.dumps({
+            "version": self.version,
+            "serviceId": self.service_id,
+            "taskId": self.task_id,
+            "maxAvailability": self.max_availability,
+            "currentAvailability": self.current_availability,
+            "requestType": self.request_type.value  # Using .value for Enum serialization
+        }, indent=2)
+
+    @classmethod
+    def from_json(cls, json_str: str) -> 'ScalingRequest':
+        """Creates a ScalingRequest from a JSON string"""
+        data = json.loads(json_str)
+        return cls(
+            service_id=data["serviceId"],
+            task_id=data["taskId"],
+            max_availability=data["maxAvailability"],
+            current_availability=data["currentAvailability"],
+            request_type=ScalingRequestAlert(data["requestType"])  # Convert string to Enum
+        )
diff --git a/amqp/model/test_snowflake_id.py b/amqp/model/test_snowflake_id.py
index 545cebf..5027d12 100644
--- a/amqp/model/test_snowflake_id.py
+++ b/amqp/model/test_snowflake_id.py
@@ -1,26 +1,26 @@
 from unittest import TestCase
 
-from amqp.adapter.amq_message_factory import AMQMessageFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.model.snowflake_id import SnowflakeId
 
 
 class SnowflakeIdTest(TestCase):
 
     def test_equals(self):
-        _id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
-        _id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
+        _id_1 = DataMessageFactory.get_instance(1).generate_snowflake_id()
+        _id_2 = DataMessageFactory.get_instance(1).generate_snowflake_id()
         self.assertFalse(_id_1 == _id_2)
         _id_3 = SnowflakeId(_id_1.get_bytes())
         self.assertTrue(_id_1 == _id_3)
 
     def test_lt(self):
-        _id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
-        _id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
+        _id_1 = DataMessageFactory.get_instance(1).generate_snowflake_id()
+        _id_2 = DataMessageFactory.get_instance(1).generate_snowflake_id()
         self.assertTrue(_id_1 < _id_2)
 
 
     def test_from_hex(self):
-        _factory = AMQMessageFactory.get_instance(1)
+        _factory = DataMessageFactory.get_instance(1)
         _id_1 = _factory.generate_snowflake_id()
         _id_1_ser = _id_1.as_string()
         _id_2 = SnowflakeId.from_hex(_id_1_ser)
@@ -31,18 +31,18 @@ class SnowflakeIdTest(TestCase):
                         in str(context.exception))
 
     def test_get_bytes(self):
-        _factory = AMQMessageFactory.get_instance(1)
+        _factory = DataMessageFactory.get_instance(1)
         _id_1 = _factory.generate_snowflake_id()
         _id_1_ser = _id_1.get_bytes()
         self.assertEqual(SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1_ser))
 
     def test_get_bits(self):
-        _factory = AMQMessageFactory.get_instance(1)
+        _factory = DataMessageFactory.get_instance(1)
         _id_1 = _factory.generate_snowflake_id()
         self.assertEqual(2, len(_id_1.get_bits()))
 
 
     def test_as_string(self):
-        _factory = AMQMessageFactory.get_instance(1)
+        _factory = DataMessageFactory.get_instance(1)
         _id_1 = _factory.generate_snowflake_id()
         self.assertEqual(2 * SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1.as_string()))
diff --git a/amqp/model/uploaded_file.py b/amqp/model/uploaded_file.py
deleted file mode 100644
index 4d14346..0000000
--- a/amqp/model/uploaded_file.py
+++ /dev/null
@@ -1,21 +0,0 @@
-import io
-
-
-class UploadedFile:
-    def __init__(self, filename: str = None):
-        self.filename = filename
-        self.body_b64 = io.BytesIO()
-
-
-def from_bytes(input_bytes: bytes) -> UploadedFile:
-    """
-    Builds the UploadedFile from its serialized (byte array) form.
-    :param input_bytes: serialized message
-    :return: UploadedFile instance
-    """
-    prolog_len = (input_bytes[0] << 8) + input_bytes[1]
-    metadata = input_bytes[2:prolog_len + 2].decode()
-    body_b64 = input_bytes[prolog_len + 2:]
-
-    return UploadedFile()
-
diff --git a/amqp/rabbitmq/rabbit_mq_client.py b/amqp/rabbitmq/rabbit_mq_client.py
new file mode 100644
index 0000000..0e1a0ea
--- /dev/null
+++ b/amqp/rabbitmq/rabbit_mq_client.py
@@ -0,0 +1,174 @@
+import logging
+
+import aio_pika
+from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
+from pika.exchange_type import ExchangeType
+
+from amqp.adapter.data_message_factory import DataMessageFactory
+from amqp.model.model import DataMessage, AMQRoute
+from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE, DLQ_EXCHANGE, AMQ_RPC_EXCHANGE
+from amqp.router.router_consumer import RouterConsumer
+from amqp.router.utils import sanitize_as_rabbitmq_name
+
+
+class RabbitMQClient:
+    PREFETCH_COUNT = 1
+    DLQ_ARGS = {"x-dead-letter-exchange": DLQ_EXCHANGE}
+
+    def __init__(self, configuration):
+        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 = configuration
+        logging.info("*******************************************")
+        logging.info("******** RabbitMQProducer.init(): %s",
+                     self.amq_configuration.dispatch.amq_host)
+        logging.info("*******************************************")
+        self.router = RouterConsumer(configuration)
+        # this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+
+
+        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)
+        await self.router.init_routing_paths_consumer(self.router.channel)
+        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=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(" [*] RabbitMQClient.cleanup()")
+        self.router.cleanup()  # de-register route(s)
+        if self.router.is_open(self.channel):
+            try:
+                self.channel.close()
+                if not self.connection.is_closed:
+                    self.connection.close()
+            except Exception as e:
+                logging.error(" [E] RabbitMQClient.PreDestroy cleanup - connection close error: %s", e)
+
+    async def register_inbound_routes(self, route_mapping: str, callback):
+        """
+        Register inbound routes for the AMQ adapter. The route mapping is a comma-separated string that contains
+        list of routes. A route is the mapping of domain and path to the exchange and queue name.
+        This method declare (and thus create) the queue and exchange, bind them together using the
+        regex-like routing expression given in the route.
+        After this the Service is listening on (consuming from) the queue for incoming messages.
+        """
+        if self.router.is_open(self.channel):
+            queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
+            logging.info(" [*] RabbitMQClient register routes, cfg mapping: [%s]", route_mapping)
+            # convert the comma-separated string into list of AMQRoute objects
+            requested_routes = self.router.unwrap_route_list(route_mapping)
+            for route in requested_routes:
+                try:
+                    # Note: AMQRoute allows one of the exchange or queue to be None, in which case the default is used
+                    queue_name = route.queue if route.queue else self.router.get_consume_queue_name()
+                    logging.info(" [*] RabbitMQClient register route: [%s]", route)
+                    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
+                    )
+                    await queue.bind(exchange, route.key)
+                    await queue.consume(callback, no_ack=False)
+                    # Register the route with the router and publish its detail to other adapters
+                    await self.router.register_route(
+                        AMQRoute(
+                            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 Exception as err:
+                    logging.error(" [E] Callback registration INCOMPLETE, %s", err)
+        else:
+            logging.warning(" [W] Channel is null, cannot register routes.")
+
+    async def register_data_reply_callback(self, rpc_callback):
+        await self.reply_to_queue.consume(callback=rpc_callback, no_ack=False)
+
+    async def setup_amq_channel(self, loop):
+        """
+        Set up the RabbitMQ connection
+        :return: nothing, it applies changes to instance variables
+        """
+        self.connection = await aio_pika.connect_robust(
+            host=self.amq_configuration.dispatch.amq_host,
+            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
+        )
+        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_internal_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))
+
+    async def send_message(self, message: DataMessage):
+        """
+        The API routine that sends a datagram message (DataMessage) to the destination. The Destination
+        is determined from the DataMessage metadata section, using `domain` and `path` as key.
+        :param message: message to send
+        :return: nothing (void) as submitting message to RabbitMQ exchange does not return anything
+         meaningful. There will be asynch ACK later that will confirm the message was sent to RMQ.
+        """
+        try:
+            route = self.router.find_route_by_message(message)
+            if route and self.rpc_exchange:
+                if route.timeout <= 0:
+                    # No response expected
+                    pika_message: aio_pika.message.Message = aio_pika.message.Message(
+                        body=DataMessageFactory.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
+                    pika_message: aio_pika.message.Message = aio_pika.message.Message(
+                        body=DataMessageFactory.serialize(message),
+                        content_encoding='utf-8',
+                        delivery_mode=2,
+                        correlation_id=str(message.id),
+                        reply_to=self.router.get_reply_to_queue_name()
+                    )
+                    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())
+            else:
+                logging.warning("Message not sent, no route for %s/%s", message.domain, message.path)
+
+        except Exception as err:
+            logging.error("Send message failed: %s", err)
+
+    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)
diff --git a/amqp/rabbitmq/rabbit_mq_consumer.py b/amqp/rabbitmq/rabbit_mq_consumer.py
deleted file mode 100644
index 3963074..0000000
--- a/amqp/rabbitmq/rabbit_mq_consumer.py
+++ /dev/null
@@ -1,81 +0,0 @@
-import logging
-
-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
-from amqp.router.router_consumer import RouterConsumer
-from amqp.router.utils import sanitize_as_rabbitmq_name
-
-
-class RabbitMQConsumer(RabbitMQProducer):
-    PREFETCH_COUNT = 1
-    DLQ_ARGS = {"x-dead-letter-exchange": DLQ_EXCHANGE}
-
-    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)
-        await self.router.init_routing_paths_consumer(self.router.channel)
-        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=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()")
-        self.router.cleanup()  # de-register route(s)
-        if self.router.is_open(self.channel):
-            try:
-                self.channel.close()
-            except Exception as e:
-                logging.error(" [E] RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
-
-    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
-            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)
-                    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
-                    )
-                    await queue.bind(exchange, route.key)
-                    await queue.consume(callback, no_ack=False)
-                    await self.router.register_route(
-                        AMQRoute(
-                            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 Exception as err:
-                    logging.error(" [E] Callback registration INCOMPLETE, %s", err)
-        else:
-            logging.warning(" [W] Channel is null, cannot register routes.")
-
-    async def register_rpc_handler(self, rpc_consumer):
-        await self.reply_to_queue.consume(callback=rpc_consumer,
-            no_ack=False
-        )
diff --git a/amqp/rabbitmq/rabbit_mq_producer.py b/amqp/rabbitmq/rabbit_mq_producer.py
deleted file mode 100644
index d67db11..0000000
--- a/amqp/rabbitmq/rabbit_mq_producer.py
+++ /dev/null
@@ -1,104 +0,0 @@
-import logging
-import aio_pika
-
-from amqp.adapter.amq_message_factory import AMQMessageFactory
-from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQMessage, AMQRoute
-from amqp.router.router_producer import RouterProducer
-
-
-class RabbitMQProducer:
-    PREFETCH_COUNT = 1
-
-    def __init__(self, amq_configuration: AMQConfiguration, router: RouterProducer = 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",
-                     self.amq_configuration.dispatch.amq_host)
-        logging.info("*******************************************")
-        self.router = router or RouterProducer(amq_configuration)
-        # this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
-
-    async def init(self, loop):
-        await self.setup_amq_channel(loop)
-
-    def cleanup(self):
-        if self.connection:
-            try:
-                self.connection.close()
-            except Exception as e:
-                logging.error("RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
-
-    async def setup_amq_channel(self, loop):
-        """
-        Set up the RabbitMQ connection
-        :return: nothing, it applies changes to instance variables
-        """
-        self.connection = await aio_pika.connect_robust(
-            host=self.amq_configuration.dispatch.amq_host,
-            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
-        )
-        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))
-
-    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.
-        :param message: message to send
-        :return: nothing (void) as submitting message to RabbitMQ exchange does not return anything
-         meaningful. There will be asynch ACK later that will confirm the message was sent to RMQ.
-        """
-        try:
-            route = self.router.find_route_by_message(message)
-            if route and self.rpc_exchange:
-                if route.timeout <= 0:
-                    # No response expected
-                    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
-                    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()
-                    )
-                    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())
-            else:
-                logging.warning("Message not sent, no route for %s/%s", message.domain, message.path)
-
-        except Exception as err:
-            logging.error("Send message failed: %s", err)
-
-    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)
diff --git a/amqp/router/router_base.py b/amqp/router/router_base.py
index 2c58bd9..d5a3426 100644
--- a/amqp/router/router_base.py
+++ b/amqp/router/router_base.py
@@ -2,7 +2,7 @@ import logging
 from typing import Callable, List, Set
 
 from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE
-from amqp.model.model import AMQRoute, AMQMessage
+from amqp.model.model import AMQRoute, DataMessage
 from amqp.router.route_database import RouteDatabase
 from amqp.router.utils import sanitize_as_rabbitmq_name
 
@@ -26,7 +26,7 @@ class RouterBase:
         self.route_added_notifier: Callable[[AMQRoute], None] | None = None
         self.route_removed_notifier: Callable[[AMQRoute], None] | None = None
 
-    def get_routing_key(self, message: AMQMessage) -> str:
+    def get_routing_key(self, message: DataMessage) -> str:
         return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
 
     def wrap_routes(self) -> str:
@@ -46,7 +46,7 @@ class RouterBase:
     def find_route(self, domain: str, path: str) -> AMQRoute | None:
         return self.route_database.find_route(domain, path)
 
-    def find_route_by_message(self, message: AMQMessage) -> AMQRoute | None:
+    def find_route_by_message(self, message: DataMessage) -> AMQRoute | None:
         return self.route_database.find_route(message.domain, message.path)
 
     def add_routes(self, message: str) -> None:
diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py
index 0cd6709..648476f 100644
--- a/amqp/router/router_consumer.py
+++ b/amqp/router/router_consumer.py
@@ -4,7 +4,7 @@ from typing import Set
 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
+from amqp.model.model import ServiceMessage, AMQRoute
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import CM_C_REQ_QUEUE, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, ANY_RECIPIENT
 from amqp.router.router_producer import RouterProducer
@@ -61,7 +61,7 @@ class RouterConsumer(RouterProducer):
 
         await message.ack(False)
 
-        cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
+        cm_message: ServiceMessage = 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(cm_message), route_mapping)
@@ -69,7 +69,7 @@ class RouterConsumer(RouterProducer):
         if cm_message.is_valid() and cm_message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping:
             if cm_message.reply_to != self.amq_configuration.amq_adapter.service_name:
                 try:
-                    service_message: CleverMicroServiceMessage = self.service_message_factory.of(
+                    service_message: ServiceMessage = self.service_message_factory.of(
                         ServiceMessageType.ROUTING_DATA, route_mapping, cm_message.reply_to
                     )
                     await self.publish_service_message(service_message, self.cm_response_exchange)
@@ -90,7 +90,7 @@ class RouterConsumer(RouterProducer):
         """
         try:
             if self.is_open(self.channel) and route.queue:
-                serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
+                serviceMessage: ServiceMessage = self.service_message_factory.of(
                     ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT
                 )
                 await self.publish_service_message(serviceMessage, self.cm_response_exchange)
@@ -110,7 +110,7 @@ class RouterConsumer(RouterProducer):
         """
         logging.info(" [DBG] RouterConsumer.removeRoute, route: %s", route.as_string())
         try:
-            service_message: CleverMicroServiceMessage = self.service_message_factory.of(
+            service_message: ServiceMessage = self.service_message_factory.of(
                 ServiceMessageType.ROUTING_DATA_REMOVE,
                 route.as_string(),
                 ANY_RECIPIENT
diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py
index 251be28..dff385c 100644
--- a/amqp/router/router_producer.py
+++ b/amqp/router/router_producer.py
@@ -13,9 +13,9 @@ from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractRob
     AbstractIncomingMessage
 from pika.exchange_type import ExchangeType
 
-from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
+from amqp.adapter.service_message_factory import ServiceMessageFactory
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import CleverMicroServiceMessage
+from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import RouterBase, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, CM_C_AMQ_QUEUE, \
     CM_P_RESP_QUEUE, CM_P_REPLY_TO, ANY_RECIPIENT
@@ -30,7 +30,7 @@ class RouterProducer(RouterBase):
          """
         super().__init__()
         self.connection: AbstractRobustConnection | None = None
-        self.service_message_factory = CleverMicroServiceMessageFactory(configuration.amq_adapter.service_name)
+        self.service_message_factory = ServiceMessageFactory(configuration.amq_adapter.service_name)
         self.amq_configuration: AMQConfiguration = configuration
         self.channel: AbstractRobustChannel | None = None
         self.cm_request_exchange: AbstractRobustExchange | None = None
@@ -40,9 +40,9 @@ class RouterProducer(RouterBase):
                      self.amq_configuration.amq_adapter.service_name,
                      self.amq_configuration.amq_adapter.route_mapping)
 
-    async def init_routing_paths(self, channel: AbstractRobustChannel | None):
+    async def init_internal_routing_paths(self, channel: AbstractRobustChannel | None):
         """
-         * Initialize router according to Producer mode.
+         * Initialize internal routes for ServiceMessages.
          * The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues
          * created to enable the service discovery.
          * The Producer mode sends out routing data request on CleverMicroRequest and listens
@@ -94,7 +94,7 @@ class RouterProducer(RouterBase):
               """
         # ACK message now so that it is not being redelivered
         await message.ack(multiple=False)
-        cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
+        cm_message: ServiceMessage = 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",
                      message.delivery_tag,
@@ -114,7 +114,7 @@ class RouterProducer(RouterBase):
          """
         logging.info(" [*] RouterProducer.requestRoutesFromAdapters, queue name: %s", CM_REQUEST_EXCHANGE)
         try:
-            serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
+            serviceMessage: ServiceMessage = self.service_message_factory.of(
                 ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
             )
             if not await self.publish_service_message(serviceMessage, self.cm_request_exchange):
@@ -125,7 +125,7 @@ class RouterProducer(RouterBase):
             logging.error(" [E] RouterProducer failed request routing data: %s", ioe)
 
     async def publish_service_message(self,
-                                      message: CleverMicroServiceMessage,
+                                      message: ServiceMessage,
                                       exchange: AbstractRobustExchange) -> bool:
         """
          * Publishes a service message to the service queue and logs success log entry.
diff --git a/amqp/router/serializer.py b/amqp/router/serializer.py
index 8a916c7..1543802 100644
--- a/amqp/router/serializer.py
+++ b/amqp/router/serializer.py
@@ -1,3 +1,7 @@
+"""
+Helper functions used to aid with (de)serialization of messages sent over AMQP.
+"""
+
 import struct
 from io import BytesIO
 from typing import List, Union, Dict
diff --git a/amqp/router/test_route_database.py b/amqp/router/test_route_database.py
index 5b4ac03..e9e1df9 100644
--- a/amqp/router/test_route_database.py
+++ b/amqp/router/test_route_database.py
@@ -2,7 +2,7 @@ from unittest import TestCase
 
 import pytest
 
-from amqp.adapter.amq_message_factory import AMQMessageFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.model.model import AMQRoute
 from amqp.router.route_database import RouteDatabase
@@ -75,7 +75,7 @@ class TestRouteDatabase(TestCase):
         self.assertTrue(ROUTE_2 in wrap)
 
     def test_find_route(self):
-        message = AMQMessageFactory.get_instance(111).create_request_message(
+        message = DataMessageFactory.get_instance(111).create_request_message(
               "POST", "clever3-book.localhost",
               "/aa/bb?cc=d", {}, ""
         )
diff --git a/amqp/router/test_router_base.py b/amqp/router/test_router_base.py
index c145e07..51a9bf7 100644
--- a/amqp/router/test_router_base.py
+++ b/amqp/router/test_router_base.py
@@ -3,7 +3,7 @@ from unittest import TestCase
 
 import pytest
 
-from amqp.adapter.amq_message_factory import AMQMessageFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.model.model import AMQRoute
 from amqp.router.router_base import RouterBase
@@ -23,7 +23,7 @@ class TestRouterBase(TestCase):
     @pytest.fixture(autouse=True)
     def setup_each_test(self):
         self.base = RouterBase()
-        self.factory = AMQMessageFactory.get_instance(111)
+        self.factory = DataMessageFactory.get_instance(111)
         if len(self.base.get_routes()) == 0:
             r1 = AMQRouteFactory.from_string(ROUTE_1)
             r2 = AMQRouteFactory.from_string(ROUTE_2)
@@ -79,7 +79,7 @@ class TestRouterBase(TestCase):
         self.assertIsNone(_route)
 
     def test_find_route_by_message(self):
-        message = AMQMessageFactory.get_instance(111).create_request_message(
+        message = DataMessageFactory.get_instance(111).create_request_message(
               "POST", "clever3-book.localhost",
               "/aa/bb?cc=d", {}, ""
         )
diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py
new file mode 100644
index 0000000..cf9ac6d
--- /dev/null
+++ b/amqp/service/amq_message_handler.py
@@ -0,0 +1,332 @@
+import asyncio
+import logging
+import sys
+import threading
+import time
+import concurrent.futures
+import os
+import traceback
+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.propagation.tracecontext import TraceContextTextMapPropagator
+from opentelemetry.trace import Tracer, TraceState
+
+from amqp.adapter.data_message_factory import DataMessageFactory
+from amqp.adapter.data_response_factory import DataResponseFactory
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
+from amqp.adapter.file_downloader import on_message
+from amqp.adapter.service_message_factory import ServiceMessageFactory
+from amqp.config.amq_configuration import AMQConfiguration
+from amqp.model.model import DataResponse, DataMessage, MultipartDataMessage, ScalingRequest, ScalingRequestAlert
+from amqp.model.service_message_type import ServiceMessageType
+from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
+from amqp.router.router_base import ANY_RECIPIENT
+from amqp.router.serializer import map_as_string
+
+# Track the number of messages currently being processed
+current_parallel_executions = 0
+last_data_message_time = 0 # helps detect IDLE condition
+last_backpressure_event_time = 0 # helps prevent flooding the system with backpressure events
+last_backpressure_event = ScalingRequestAlert.IDLE
+
+# Global lock to synchronize access to the current_parallel_executions variable
+lock = threading.Lock()
+parallel_workers = int(os.getenv("PARALLEL_WORKERS", default=0))
+swarm_service_id = os.getenv("SWARM_SERVICE_ID", default="clever-amqp-service")
+swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter")
+# Shared thread pool for parallel execution
+executor = concurrent.futures.ThreadPoolExecutor(max_workers=parallel_workers)
+
+
+def collect_trace_info():
+    trace_map = {}
+    TraceContextTextMapPropagator().inject(
+        carrier=trace_map,
+        context=trace.context_api.get_current()
+    )
+    return trace_map
+
+
+def get_default_trace_state():
+    return TraceState(entries={})
+
+
+def set_text(carrier, key, value):
+    carrier[key] = value
+
+
+class DataMessageHandler:
+    def __init__(self,
+                 service_adapter: CleverThisServiceAdapter,
+                 tracer: Tracer,
+                 message_factory: DataMessageFactory,
+                 service_message_factory: ServiceMessageFactory,
+                 reply_to_exchange: AbstractRobustExchange,
+                 rabbit_mq_client: RabbitMQClient,
+                 amq_configuration: AMQConfiguration,
+                 loop):
+        self.service_adapter: CleverThisServiceAdapter = service_adapter
+        self.tracer = tracer
+        self.message_factory: DataMessageFactory = message_factory
+        self.service_message_factory: ServiceMessageFactory = service_message_factory
+        self.reply_to_exchange: AbstractRobustExchange = reply_to_exchange
+        self.rabbit_mq_client: RabbitMQClient = rabbit_mq_client
+        self.rabbit_mq_url = amq_configuration.dispatch.rabbit_mq_url
+        self.output_dir = amq_configuration.dispatch.download_dir
+        self.loop = loop
+        # Dictionary to store outstanding Futures
+        self.outstanding: Dict[str, asyncio.Future] = {}
+        self.reply_to: str | None = None
+        self.backpressure_monitor_window = amq_configuration.backpressure.time_window / 1000  # convert to seconds
+
+        global parallel_workers
+        _backpressure_treshold = amq_configuration.backpressure.threshold
+        """
+        Determine the number of parallel workers based on the backpressure threshold and PARALLEL_WORKERS env var.
+        If env var PARALLEL_WORKERS is not set (or set to 0), use the backpressure threshold config
+        as the number of parallel workers.
+        """
+        if _backpressure_treshold > 0:
+            if parallel_workers <= 0:
+                parallel_workers = _backpressure_treshold
+
+    async def monitor_backpressure(self):
+        while True:
+            global last_data_message_time
+            global last_backpressure_event_time
+            global last_backpressure_event
+
+            now = time.time()
+            event: ScalingRequestAlert = ScalingRequestAlert.UPDATE
+            if now - last_backpressure_event_time >= self.backpressure_monitor_window:
+                if current_parallel_executions == 0 and now - last_data_message_time > self.backpressure_monitor_window:
+                    event = ScalingRequestAlert.IDLE
+                elif current_parallel_executions > parallel_workers - 2:
+                    event = ScalingRequestAlert.OVERLOAD
+
+            last_backpressure_event_time = now
+            last_backpressure_event = event
+            last_backpressure_event_time = now
+            logging.info(f"Backpressure event: {event}, threads: {current_parallel_executions} of {parallel_workers}")
+
+            scaling_request: ScalingRequest = ScalingRequest(
+                swarm_service_id, swarm_task_id,
+                parallel_workers, parallel_workers - current_parallel_executions,
+                event
+            )
+            # Address the message to any adapter capable of supporting BACKPRESSURE request
+            scale_up_request = self.service_message_factory.of(
+                ServiceMessageType.BACKPRESSURE,
+                scaling_request.to_json(),
+                self.reply_to if self.reply_to else ANY_RECIPIENT
+            )
+            asyncio.run_coroutine_threadsafe(
+                self.rabbit_mq_client.router.publish_service_message(
+                    scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange
+                ),
+                loop=self.loop
+            )
+            await asyncio.sleep(self.backpressure_monitor_window)  # sleep in seconds
+
+    def sync_on_message(self,
+                        message: AbstractIncomingMessage,
+                        amq_message: DataMessage,
+                        loop: AbstractEventLoop) -> DataResponse:
+        """
+        Synchronous version of on_message method. This is a wrapper around the async on_message method.
+        It also counts parallel executions and handles backpressure.
+        :param message: RabbitMQ raw message
+        :param amq_message: DataMessage
+        :param loop: Event loop
+        :return: DataResponse
+        """
+        global current_parallel_executions
+
+        # Increment the counter for parallel executions
+        with lock:
+            current_parallel_executions += 1
+            logging.info(f"PARALLEL: Current Capacity [{current_parallel_executions}] of [{parallel_workers}]")
+            if message.reply_to:
+                self.reply_to = message.reply_to
+            if current_parallel_executions > parallel_workers - 2:
+                logging.warning("Warning: Capacity close to depleted!")
+                # Send a scaling request to the Management service
+                scaling_request: ScalingRequest = ScalingRequest(
+                    swarm_service_id, swarm_task_id,
+                    parallel_workers, parallel_workers - current_parallel_executions,
+                    ScalingRequestAlert.OVERLOAD
+                )
+                # Address the message to any adapter capable of supporting BACKPRESSURE request
+                scale_up_request = self.service_message_factory.of(
+                    ServiceMessageType.BACKPRESSURE,
+                    scaling_request.to_json(),
+                    self.reply_to if self.reply_to else ANY_RECIPIENT
+                )
+                asyncio.run_coroutine_threadsafe(
+                    self.rabbit_mq_client.router.publish_service_message(
+                        scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange
+                    ),
+                    loop=loop
+                )
+                # update / reset time-window so that the OVERLOAD is not sent too often
+                global last_data_message_time
+                last_data_message_time = time.time()
+
+        # Call the async on_message method. This shall also ACK the message after processing.
+        response: DataResponse = asyncio.run(self.service_adapter.on_message(amq_message))
+        try:
+            logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s",
+                         message.delivery_tag, message.reply_to, response.id)
+            asyncio.run_coroutine_threadsafe(self.send_reply(message.reply_to, response), loop=loop)
+        except Exception as e:
+            logging.info(f"[E] Error processing message: {e}")
+            tb = traceback.extract_tb(sys.exc_info()[2])[-1]
+            logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {e}")
+            asyncio.run_coroutine_threadsafe(message.nack(requeue=False), loop=loop)
+
+        if isinstance(amq_message, MultipartDataMessage):
+            # remove the downloaded files from the output_dir
+            for file_id, filename in amq_message.extra_data.items():
+                try:
+                    logging.info(f" [*][{message.delivery_tag}] Deleting file: {file_id} in {filename}")
+                    os.remove(filename)
+                except OSError as e:
+                    logging.error(f"Error deleting file {filename}: {e}")
+
+        with lock:
+            current_parallel_executions -= 1
+
+        return response
+
+
+    async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
+        """
+        Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object
+        Ensures Jaeger trace-id propagation.
+        Invokes custom handler/mapper/adapter method that corresponds to this message
+        :param message:
+        :return:
+        """
+        start = time.time()
+        delivery_tag = message.delivery_tag
+        debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {message.headers}, props: {message.properties}"
+        logging.debug(debug)
+        addressing: int = message.headers.get('clevermicro.addressing', 0)
+
+        amq_message: DataMessage | None = None
+        if addressing == 0:
+            amq_message = DataMessageFactory.from_bytes(message.body)
+        else:
+            loop = asyncio.get_event_loop()
+            amq_message = await DataMessageFactory.from_stream(message.body, rabbitmq_url=self.rabbit_mq_url, loop=loop)
+            logging.info(f" [DBG] ######[{delivery_tag}] AMQ Message from stream: {amq_message}")
+            if amq_message is not None:
+                extra_data = await on_message(
+                    message=message,
+                    json_data=amq_message.body(),
+                    rabbitmq_url=self.rabbit_mq_url,
+                    loop=loop,
+                    output_dir=self.output_dir)
+                logging.info(f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}")
+                amq_message = MultipartDataMessage(amq_message, extra_data)
+
+        if amq_message is not None:
+            logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s",
+                         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())
+            self.tracer.start_span(name="AMQ-Adapter-Consumer", context=context)
+
+            logging.info(" [*] CTX=%s", context)
+            # current_span = self.tracer.start_span("AMQ-Adapter.Processing.Request", parent=parent_span)
+            # context_with_span = trace.set_span_in_context(current_span, context)
+            # logging.info(" [*] CTX2=%s", context_with_span)
+            # propagator.inject(context_with_span, amq_message.trace_info, self.set_text)
+            # logging.info(" [***] NEW TI=%s CTX=%s SCOPE=%s",
+            #             map_as_string(amq_message.trace_info), context_with_span, context)
+
+            if amq_message.magic == DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
+                future = executor.submit(self.sync_on_message, message, amq_message, self.loop)
+                # response: DataResponse = asyncio.run(self.http_adapter.on_message(amq_message))
+                # Wait  and send it back to the reply queue
+                #try:
+                #    response: DataResponse = future.result()  # Block until the result is available
+                #    await self.send_reply(message.reply_to, delivery_tag, response)
+                #except Exception as e:
+                #    logging.info(f"[E] Error processing message: {e}")
+                #    tb = traceback.extract_tb(sys.exc_info()[2])[-1]
+                #    logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {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
+        else:
+            logging.error(" [E] ######[%s] No valid AMQ Message present.", delivery_tag)
+
+        global last_data_message_time
+        last_data_message_time = time.time()
+        logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, last_data_message_time - start)
+        await message.ack(multiple=False)
+
+    async def send_reply(self, reply_to: str, response: DataResponse):
+        """
+        The Service side code. When the CleverXXX service has output ready, in DataResponse object,
+        call this method to return the reply back to the caller
+        :param reply_to:
+        :param response:
+        :return:
+        """
+        if reply_to:
+            pika_message: Message = Message(
+                body=DataResponseFactory.serialize(response),
+                correlation_id=response.id,
+                content_type=response.content_type[0]
+                if isinstance(response.content_type, list) else response.content_type
+            )
+            await self.reply_to_exchange.publish(
+                message=pika_message,
+                routing_key=reply_to
+            )
+
+    async def reply_received_callback(self, message: AbstractIncomingMessage):
+        """
+        The reply for DataMessage that we sent out earlier - must match the received response with
+        the original request and respond to the caller.
+        :param message:
+        :return:
+        """
+        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
+        logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}")
+
+        future: Future = self.outstanding.pop(reply_id, None)
+        # if reply is None:
+        # logging.warn(" [W] ######[%s] IGNORING REPLY. No outstanding request for ID %s",
+        #             delivery_tag, reply_id)
+        # else:
+        # result = DataResponseFactory.from_bytes(body)
+        # reply.tracing_span().end()
+        # TODO - figure out the right response format (not needed for Clever Swarm v0.1)
+        # that currently runs Java version, this method here is for the future compatibility to enable
+        # 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: DataResponse = DataResponseFactory.from_bytes(message.body)
+            # Complete the Future with the response data
+            future.set_result(response.body())
+
+        await message.ack(multiple=False)
diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py
index 8f2211f..0adcedd 100644
--- a/amqp/service/amq_service.py
+++ b/amqp/service/amq_service.py
@@ -6,14 +6,14 @@ from asyncio import Future
 
 from aio_pika.abc import AbstractRobustExchange
 
-from amqp.adapter.amq_message_factory import AMQMessageFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.amq_route_factory import AMQRouteFactory
-from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
+from amqp.adapter.service_message_factory import ServiceMessageFactory
 
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQMessage
-from amqp.rabbitmq.rabbit_mq_consumer import RabbitMQConsumer
-from amqp.service.service_message_handler import AMQServiceMessageHandler
+from amqp.model.model import DataMessage
+from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
+from amqp.service.amq_message_handler import DataMessageHandler
 from amqp.service.tracing import initialize_trace
 
 
@@ -38,18 +38,18 @@ class AMQService:
         self.loop = asyncio.get_event_loop()
 
         self.amq_configuration = amq_configuration
-        self.message_factory = AMQMessageFactory.get_instance(self.amq_configuration.amq_adapter.generator_id)
-        self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration)
+        self.message_factory = DataMessageFactory.get_instance(self.amq_configuration.amq_adapter.generator_id)
+        self.rabbit_mq_client = RabbitMQClient(self.amq_configuration)
         self.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name)
-        self.amq_service_message_handler = None
-        self.service_message_factory = CleverMicroServiceMessageFactory(
+        self.amq_data_message_handler = None
+        self.service_message_factory = ServiceMessageFactory(
             self.amq_configuration.amq_adapter.service_name
         )
         # self.init(loop) will initialize RabbitMQ connection
         self.loop.run_until_complete(self.init(self.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()
+        # self.consumer_thread = asyncio.create_task(self.rabbit_mq_client.channel.start_consuming())
+        # self.rabbit_mq_client.request_routes()
 
         logging.info("***********************************************")
         logging.info("******  AMQ   Service  Init  Complete *********")
@@ -57,40 +57,41 @@ class AMQService:
         self.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(
+        exchange: AbstractRobustExchange = await self.rabbit_mq_client.init(loop)
+        self.amq_data_message_handler = DataMessageHandler(
             service_adapter,
             self.tracer,
             self.message_factory,
+            self.service_message_factory,
             exchange,
-            self.rabbit_mq_consumer.channel,
-            self.amq_configuration.dispatch.rabbit_mq_url,
-            output_dir=self.amq_configuration.dispatch.download_dir
+            self.rabbit_mq_client,
+            self.amq_configuration,
+            loop=loop
         )
         await self.register_routes()
-        await self.rabbit_mq_consumer.request_routes()
+        await self.rabbit_mq_client.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()
+        logging.info(" [DBG] AMQService.cleanup() rmqc=%s", self.rabbit_mq_client)
+        if self.rabbit_mq_client:
+            self.rabbit_mq_client.cleanup()
 
     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)
-            await self.rabbit_mq_consumer.init(loop=self.loop)
+        if not self.rabbit_mq_client.channel or self.rabbit_mq_client.channel.is_closed:
+            self.rabbit_mq_client = RabbitMQClient(self.amq_configuration)
+            await self.rabbit_mq_client.init(loop=self.loop)
             await self.register_routes()
-            await self.rabbit_mq_consumer.request_routes()
+            await self.rabbit_mq_client.request_routes()
 
     async def register_routes(self) -> AbstractRobustExchange | None:
         route_mapping = self.amq_configuration.amq_adapter.route_mapping
         if AMQRouteFactory.validate(route_mapping):
             try:
-                await self.rabbit_mq_consumer.register_inbound_routes(
-                    route_mapping, self.amq_service_message_handler.inbound_message_callback
+                await self.rabbit_mq_client.register_inbound_routes(
+                    route_mapping, self.amq_data_message_handler.inbound_data_message_callback
                 )
-                return await self.rabbit_mq_consumer.register_rpc_handler(
-                    self.amq_service_message_handler.reply_received_callback
+                return await self.rabbit_mq_client.register_data_reply_callback(
+                    self.amq_data_message_handler.reply_received_callback
                 )
             except Exception as e:
                 raise RuntimeError(e)
@@ -98,16 +99,16 @@ class AMQService:
             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
+    def send_message(self, message: DataMessage, future: Future):
+        self.amq_data_message_handler.outstanding[str(message.id)] = future
 
     def remove_outstanding_future(self, message_id: str):
-        self.amq_service_message_handler.outstanding.pop(message_id, None)
+        self.amq_data_message_handler.outstanding.pop(message_id, None)
 
 
 # if __name__ == "__main__":
     # amq_configuration: AMQConfiguration = AMQConfiguration('../config/application.properties.local')
     # amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAmqpAdapter(amq_configuration))
-    # amq_service.rabbit_mq_consumer.channel.start_consuming()
+    # amq_service.rabbit_mq_client.channel.start_consuming()
     # amq_service.consumer_thread.cancel()
-    # amq_service.rabbit_mq_consumer.connection.close()
+    # amq_service.rabbit_mq_client.connection.close()
diff --git a/amqp/service/delete_queues.py b/amqp/service/delete_queues.py
index ce8efe7..a0a5547 100644
--- a/amqp/service/delete_queues.py
+++ b/amqp/service/delete_queues.py
@@ -3,7 +3,7 @@ import requests
 
 def delete_queues_with_prefix_http(host='localhost', port=15672, username='springuser', password='TheCleverWho', prefix='cm-'):
     """
-    Delete queues using RabbitMQ HTTP API.
+    CLI utility. Delete queues using RabbitMQ HTTP API.
     """
     base_url = f"http://{host}:{port}/api/queues/%2F"
 
diff --git a/amqp/service/multipart_parser.py b/amqp/service/multipart_parser.py
index fab40e0..489c292 100644
--- a/amqp/service/multipart_parser.py
+++ b/amqp/service/multipart_parser.py
@@ -7,11 +7,11 @@ from fastapi import UploadFile
 from python_multipart import multipart
 from python_multipart.multipart import Field, File
 
-from amqp.model.model import AMQMessage
+from amqp.model.model import DataMessage
 
 
 class CleverMultiPartParser:
-    def __init__(self, message: AMQMessage):
+    def __init__(self, message: DataMessage):
         self.form_data = {}
         self.is_multipart = False
         self.is_json = False
diff --git a/amqp/service/service_message_handler.py b/amqp/service/service_message_handler.py
deleted file mode 100644
index 4941758..0000000
--- a/amqp/service/service_message_handler.py
+++ /dev/null
@@ -1,221 +0,0 @@
-import asyncio
-import logging
-import sys
-import time
-import concurrent.futures
-import os
-import traceback
-from asyncio import Future
-from typing import Dict
-
-from aio_pika import IncomingMessage, Message
-from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange, AbstractRobustChannel
-
-from opentelemetry import trace
-from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
-from opentelemetry.trace import Tracer, TraceState
-
-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.adapter.file_downloader import on_message
-from amqp.model.model import AMQResponse, AMQMessage, MultipartAMQMessage
-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 = {}
-    TraceContextTextMapPropagator().inject(
-        carrier=trace_map,
-        context=trace.context_api.get_current()
-    )
-    return trace_map
-
-
-def get_default_trace_state():
-    return TraceState(entries={})
-
-
-def set_text(carrier, key, value):
-    carrier[key] = value
-
-
-class AMQServiceMessageHandler:
-    def __init__(self,
-                 service_adapter: CleverThisServiceAdapter,
-                 tracer: Tracer,
-                 message_factory: AMQMessageFactory,
-                 reply_to_exchange: AbstractRobustExchange,
-                 channel: AbstractRobustChannel,
-                 rabbit_mq_url: str,
-                 output_dir: str):
-        self.service_adapter: CleverThisServiceAdapter = service_adapter
-        self.tracer = tracer
-        self.message_factory: AMQMessageFactory = message_factory
-        self.reply_to_exchange: AbstractRobustExchange = reply_to_exchange
-        self.channel: AbstractRobustChannel = channel
-        self.rabbit_mq_url = rabbit_mq_url
-        self.output_dir = output_dir
-        # Dictionary to store outstanding Futures
-        self.outstanding: Dict[str, asyncio.Future] = {}
-
-    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.
-        Invokes custom handler/mapper/adapter method that corresponds to this message
-        :param message:
-        :return:
-        """
-        start = time.time()
-        delivery_tag = message.delivery_tag
-        debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {message.headers}, props: {message.properties}"
-        logging.debug(debug)
-        addressing: int = message.headers.get('clevermicro.addressing', 0)
-
-        amq_message: AMQMessage | None = None
-        if addressing == 0:
-            amq_message = AMQMessageFactory.from_bytes(message.body)
-        else:
-            loop = asyncio.get_event_loop()
-            amq_message = await AMQMessageFactory.from_stream(message.body, rabbitmq_url=self.rabbit_mq_url, loop=loop)
-            logging.info(f" [DBG] ######[{delivery_tag}] AMQ Message from stream: {amq_message}")
-            if amq_message is not None:
-                extra_data = await on_message(
-                    message=message,
-                    json_data=amq_message.body(),
-                    rabbitmq_url=self.rabbit_mq_url,
-                    loop=loop,
-                    output_dir=self.output_dir)
-                logging.info(f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}")
-                amq_message = MultipartAMQMessage(amq_message, extra_data)
-
-        if amq_message is not None:
-            logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s",
-                         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())
-            self.tracer.start_span(name="AMQ-Adapter-Consumer", context=context)
-
-            logging.info(" [*] CTX=%s", context)
-            # current_span = self.tracer.start_span("AMQ-Adapter.Processing.Request", parent=parent_span)
-            # context_with_span = trace.set_span_in_context(current_span, context)
-            # logging.info(" [*] CTX2=%s", context_with_span)
-            # propagator.inject(context_with_span, amq_message.trace_info, self.set_text)
-            # logging.info(" [***] NEW TI=%s CTX=%s SCOPE=%s",
-            #             map_as_string(amq_message.trace_info), context_with_span, context)
-
-            if amq_message.magic == AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
-                future = executor.submit(self.service_adapter.sync_on_message, amq_message)
-                # response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message))
-                # Wait  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:
-                    logging.info(f"[E] Error processing message: {e}")
-                    tb = traceback.extract_tb(sys.exc_info()[2])[-1]
-                    logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {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
-        else:
-            logging.error(" [E] ######[%s] No valid AMQ Message present.", delivery_tag)
-
-        duration = time.time() - start
-        logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration)
-        await message.ack(multiple=False)
-
-    def on_http_response(self, delivery, deliveryTag, channel, http_response):
-        """
-        In loose coupling mode, receives the HTTP response to the REST request the HTTPAdapter made to the
-        associated CleverXXX service. Convert the response to AMQResponse message and use
-        RabbitMQ RPC reply-to mechanism to deliver this response back to API Gateway side AMQProducer.
-        :param delivery:
-        :param deliveryTag:
-        :param channel:
-        :param http_response:
-        :return:
-        """
-        response = AMQResponse(
-            delivery.properties.correlation_id,
-            http_response.status_code,
-            http_response.headers.get("Content-Type"),
-            http_response.body,
-            http_response.status_message,
-            "",
-            collect_trace_info()
-        )
-        try:
-            self.send_reply(delivery.properties.reply_to, deliveryTag, response)
-        except Exception as e:
-            tb = traceback.extract_tb(sys.exc_info()[2])[-1]
-            logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {e}")
-            raise RuntimeError(e)
-
-    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 response:
-        :return:
-        """
-        if reply_to:
-            logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s",
-                         delivery_tag, reply_to, response.id)
-            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
-            )
-            await self.reply_to_exchange.publish(
-                message=pika_message,
-                routing_key=reply_to
-            )
-
-    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 message:
-        :return:
-        """
-        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
-        logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}")
-
-        future: Future = self.outstanding.pop(reply_id, None)
-        # if reply is None:
-        # logging.warn(" [W] ######[%s] IGNORING REPLY. No outstanding request for ID %s",
-        #             delivery_tag, reply_id)
-        # else:
-        # result = AMQResponseFactory.from_bytes(body)
-        # reply.tracing_span().end()
-        # TODO - figure out the right response format (not needed for Clever Swarm v0.1)
-        # that currently runs Java version, this method here is for the future compatibility to enable
-        # 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(message.body)
-            # Complete the Future with the response data
-            future.set_result(response.body())
-
-        await message.ack(multiple=False)
diff --git a/amqp/service/stream_parser.py b/amqp/service/stream_parser.py
deleted file mode 100644
index 3e9ded1..0000000
--- a/amqp/service/stream_parser.py
+++ /dev/null
@@ -1,59 +0,0 @@
-import json
-import struct
-import io
-
-
-class StreamParser:
-    """
-    Parses input stream that contains metadata followed by binary data.
-    First 4 bytes are metadata length, followed by metadata in JSON format, followed by binary data.
-    The StreamParser accumulates the binary data from individual chunks and writes it to a file.
-    This can be used to parse a stream that contains a large file upload.
-    """
-    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
-        seek_to = 0
-
-    def process_chunk(self, chunk: bytes):
-        """
-        Process a chunk of data and handle it according to the rules.
-        """
-        self.buffer.write(chunk)
-        available_data = self.buffer.tell()
-        # If metadata length is not yet known, try to read it
-        if self.metadata_length is None and available_data >= 4:
-            self.buffer.seek(0)
-            self.metadata_length = struct.unpack('>I', self.buffer.read(4))[0]
-            self.seek_to = self.buffer.tell()
-
-        # 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 available_data >= 4 + self.metadata_length:
-                self.buffer.seek(self.seek_to)
-                metadata_bytes = self.buffer.read(self.metadata_length)
-                self.metadata = json.loads(metadata_bytes.decode('utf-8'))
-                self.seek_to = self.buffer.tell()
-                print("Metadata:", self.metadata)
-
-
-        # If metadata is parsed, write the remaining data to the file
-        if self.metadata is not None:
-            self.buffer.seek(self.seek_to)
-            remaining_data = self.buffer.read()
-            self.seek_to = self.buffer.tell()
-            if remaining_data:
-                if self.file_writer is None:
-                    # Prepare to write the remaining data to a file
-                    self.file_writer = open(self.metadata.get("filename", "output_file"), "wb")
-                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()
diff --git a/amqp/service/test_stream_parser.py b/amqp/service/test_stream_parser.py
deleted file mode 100644
index 65ad94b..0000000
--- a/amqp/service/test_stream_parser.py
+++ /dev/null
@@ -1,141 +0,0 @@
-import unittest
-import json
-import struct
-import os
-from unittest.mock import patch, mock_open
-
-from amqp.service.stream_parser import StreamParser
-
-
-class TestStreamParser(unittest.TestCase):
-    def setUp(self):
-        """
-        Set up the StreamParser instance for testing.
-        """
-        self.parser = StreamParser()
-
-    def tearDown(self):
-        """
-        Clean up after each test.
-        """
-        if hasattr(self.parser, 'file_writer') and self.parser.file_writer:
-            self.parser.file_writer.close()
-        if os.path.exists("test_output_file"):
-            os.remove("test_output_file")
-
-    def test_process_chunk_with_metadata_and_data(self):
-        """
-        Test processing a chunk that contains metadata and data.
-        """
-        # Create test data
-        metadata = {"filename": "test_output_file", "description": "Test file"}
-        metadata_json = json.dumps(metadata).encode('utf-8')
-        metadata_length = struct.pack('>I', len(metadata_json))
-        data = b"some binary data"
-
-        # Combine metadata length, metadata, and data into a single chunk
-        chunk = metadata_length + metadata_json + data
-
-        # Process the chunk
-        self.parser.process_chunk(chunk)
-        self.parser.close()
-
-        # Verify metadata
-        self.assertEqual(self.parser.metadata, metadata)
-        self.assertEqual(self.parser.metadata_length, len(metadata_json))
-
-        # Verify file content
-        with open("test_output_file", "rb") as f:
-            file_content = f.read()
-        self.assertEqual(file_content, data)
-
-    def test_process_chunk_in_parts(self):
-        """
-        Test processing chunks in multiple parts.
-        """
-        # Create test data
-        metadata = {"filename": "test_output_file", "description": "Test file"}
-        metadata_json = json.dumps(metadata).encode('utf-8')
-        metadata_length = struct.pack('>I', len(metadata_json))
-        data = b"some binary data"
-
-        # Split the chunk into parts
-        chunk1 = metadata_length  # First 4 bytes (metadata length)
-        chunk2 = metadata_json  # Metadata JSON
-        chunk3 = data  # Remaining data
-
-        # Process chunks one by one
-        self.parser.process_chunk(chunk1)
-        self.parser.process_chunk(chunk2)
-        self.parser.process_chunk(chunk3)
-        self.parser.close()
-
-        # Verify metadata
-        self.assertEqual(self.parser.metadata, metadata)
-        self.assertEqual(self.parser.metadata_length, len(metadata_json))
-
-        # Verify file content
-        with open("test_output_file", "rb") as f:
-            file_content = f.read()
-        self.assertEqual(file_content, data)
-
-    def test_process_chunk_without_metadata(self):
-        """
-        Test processing a chunk without metadata.
-        """
-        # Create test data (only metadata length and metadata, no additional data)
-        metadata = {"filename": "distinct_test_output_file", "description": "Test file"}
-        metadata_json = json.dumps(metadata).encode('utf-8')
-        metadata_length = struct.pack('>I', len(metadata_json))
-
-        # Combine metadata length and metadata into a single chunk
-        chunk = metadata_length + metadata_json
-
-        # Process the chunk
-        self.parser.process_chunk(chunk)
-        self.parser.close()
-
-        # Verify metadata
-        self.assertEqual(self.parser.metadata, metadata)
-        self.assertEqual(self.parser.metadata_length, len(metadata_json))
-
-        # Verify no data was written to the file
-        self.assertFalse(os.path.exists("distinct_test_output_file"))
-
-    def test_process_chunk_with_invalid_metadata(self):
-        """
-        Test processing a chunk with invalid metadata.
-        """
-        # Create invalid metadata (not JSON)
-        metadata_length = struct.pack('>I', 10)
-        invalid_metadata = b"invalid_json"
-
-        # Combine metadata length and invalid metadata into a single chunk
-        chunk = metadata_length + invalid_metadata
-
-        # Process the chunk and expect an exception
-        with self.assertRaises(json.JSONDecodeError):
-            self.parser.process_chunk(chunk)
-
-
-
-    @patch("builtins.open", new_callable=mock_open)
-    def test_file_writing(self, mock_file):
-        """
-        Test file writing functionality using a mock file.
-        """
-        # Create test data
-        metadata = {"filename": "test_output_file", "description": "Test file"}
-        metadata_json = json.dumps(metadata).encode('utf-8')
-        metadata_length = struct.pack('>I', len(metadata_json))
-        data = b"some binary data"
-
-        # Combine metadata length, metadata, and data into a single chunk
-        chunk = metadata_length + metadata_json + data
-
-        # Process the chunk
-        self.parser.process_chunk(chunk)
-
-        # Verify the file was opened and written to
-        mock_file.assert_called_once_with("test_output_file", "wb")
-        mock_file().write.assert_called_once_with(data)
diff --git a/pyproject.toml b/pyproject.toml
index 2783174..f4e63e9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -3,11 +3,12 @@ requires = ["setuptools>=61.0", "wheel"]  # Specifies build tools to use
 build-backend = "setuptools.build_meta"
 
 [tool.setuptools]
-packages = ["amqp","amqp.adapter","amqp.config","amqp.model","amqp.rabbitmq","amqp.router","amqp.service"]  # Explicitly list packages to include
+# Explicitly list packages to include`
+packages = ["amqp","amqp.adapter","amqp.config","amqp.model","amqp.rabbitmq","amqp.router","amqp.service"]
 
 [project]
 name = "amq_adapter"
-version = "0.2.10"
+version = "0.2.14"
 description = "The Python implementation of the AMQ Adaptor for CleverMicro"
 readme = "README.md"
 license = { text = "Apache License 2.0" }