feat/#63 - add iterator instead of RabbitMQ callback

amqp/adapter/data_message_factory.py

@@ -113,6 +113,25 @@ class DataMessageFactory:
             payload,
         )
 
+    def safe_rsplit(self, fq_method_name, default="_"):
+        """
+        Safely split a fully qualified method name into package, class, and method
+
+        Args:
+            fq_method_name (str): Fully qualified method name
+            default (str, optional): Default value for missing tokens. Defaults to ''.
+
+        Returns:
+            tuple: (package, class_name, method)
+        """
+        if not fq_method_name:
+            return default, default, default
+        parts = fq_method_name.rsplit(".", 2)
+        # Pad with default values if insufficient tokens
+        while len(parts) < 3:
+            parts.insert(0, default)
+        return tuple(parts)
+
     def create_rpc_message(
         self,
         fq_method_name: str,
@@ -125,7 +144,7 @@ class DataMessageFactory:
         :param args: arguments whose serialized form will form the request body
         :return: DataMessage object
         """
-        package, class_name, method = fq_method_name.rsplit(".", 2)
+        package, class_name, method = self.safe_rsplit(fq_method_name)
         headers = {"Content-Type": ["application/octet-stream"], "Return-Address": [swarm_id]}
         trace_info = {}
         body = python_type_to_json(args).encode(encoding="utf-8") if args else b""

amqp/config/amq_configuration.py

@@ -43,6 +43,11 @@ class AMQAdapter:
             self.PREFIX + "default-backpressure-enabled",
             fallback=False,
         )
+        self.callbacks_enabled = config.getboolean(
+            "CleverMicro-AMQ",
+            self.PREFIX + "callbacks-enabled",
+            fallback=False,
+        )
         # these 2 shall be provided by Docker swarm, format is a random alphanumeric string
         self.swarm_service_id = os.getenv("SWARM_SERVICE_ID", default="clever-amqp-service")
         self.swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter")

amqp/config/application.properties

@@ -20,6 +20,7 @@ cm.amq-adapter.consul-port=8500
 cm.amq-adapter.consul-max-retries=5
 cm.amq-adapter.consul-initial-retry-delay=0.2
 cm.amq-adapter.consul-id-generator-key=services/ids/counter
+cm.amq-adapter.callbacks-enabled=False
 
 # CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding
 cm.dispatch.use-dlq=true

amqp/rabbitmq/rabbit_mq_client.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Optional
 
 import aio_pika
 from aio_pika.abc import AbstractRobustExchange, AbstractRobustQueue
@@ -95,19 +96,25 @@ class RabbitMQClient:
                         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(
+                    if self.amq_configuration.amq_adapter.callbacks_enabled:
-                        name=queue_name,
+                        queue: Optional[AbstractRobustQueue] = await self.channel.declare_queue(
-                        durable=True,
+                            name=queue_name,
-                        exclusive=False,
+                            durable=True,
-                        auto_delete=False,
+                            exclusive=False,
-                        arguments=queue_args,
+                            auto_delete=False,
-                    )
+                            arguments=queue_args,
+                        )
+                    else:
+                        queue = None
                     _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)
+                    if self.amq_configuration.amq_adapter.callbacks_enabled:
-                    _c_tag = await queue.consume(callback, no_ack=False)
+                        await queue.bind(exchange, route.key)
+                        _c_tag = await queue.consume(callback, no_ack=False)
+                    else:
+                        _c_tag = None
                     # Register the route with the router and publish its detail to other adapters
                     await self.router.register_route(
                         AMQRoute(
@@ -144,6 +151,9 @@ class RabbitMQClient:
         Set up the RabbitMQ connection
         :return: nothing, it applies changes to instance variables
         """
+        logging.info(
+            f"RabbitMQClient.setupAMQ, connecting to RabbitMQ at {self.amq_configuration.dispatch.amq_host}:{self.amq_configuration.dispatch.amq_port}, loop={loop}"
+        )
         self.connection = await aio_pika.connect_robust(
             host=self.amq_configuration.dispatch.amq_host,
             port=self.amq_configuration.dispatch.amq_port,

amqp/service/amq_message_handler.py

@@ -30,6 +30,7 @@ from amqp.model.model import (
     MultipartDataMessage,
 )
 from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
+from amqp.router.utils import sanitize_as_rabbitmq_name
 
 
 def collect_trace_info():
@@ -470,7 +471,9 @@ class DataMessageHandler:
             name=rpc_exchange_name, type=ExchangeType.topic
         )
         # Publish the message to the exchange with the component name as the routing key
-        await exchange.publish(message=pika_message, routing_key=route.component_name)
+        await exchange.publish(
+            message=pika_message, routing_key=sanitize_as_rabbitmq_name(route.component_name)
+        )
 
         logging_debug(
             "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s",
@@ -514,7 +517,9 @@ class DataMessageHandler:
             name=rpc_exchange_name, type=ExchangeType.topic
         )
         # Publish the message to the exchange with the component name as the routing key
-        await exchange.publish(message=pika_message, routing_key=route.component_name)
+        await exchange.publish(
+            message=pika_message, routing_key=sanitize_as_rabbitmq_name(route.component_name)
+        )
 
         logging_info(
             "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s",

amqp/service/amq_service.py

@@ -4,7 +4,7 @@ import logging.config
 import os
 from asyncio import Future
 from threading import Thread
-from typing import Optional
+from typing import AsyncIterator, Optional
 
 from aio_pika.abc import AbstractRobustExchange
 
@@ -12,7 +12,7 @@ from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.backpressure_handler import BackpressureHandler
 from amqp.adapter.consul_global_id_generator import get_unique_instance_id
 from amqp.adapter.data_message_factory import DataMessageFactory
-from amqp.adapter.logging_utils import logging_info, logging_warning
+from amqp.adapter.logging_utils import logging_error, logging_info, logging_warning
 from amqp.adapter.service_message_factory import ServiceMessageFactory
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import AMQRoute, DataMessage
@@ -65,6 +65,7 @@ class AMQService:
         self.backpressure_thread: Optional[Thread] = None
         self.backpressure_handler: Optional[BackpressureHandler] = None
         self.maximum_availability = -1
+        self.future: Future = asyncio.Future()
 
     # =======================================================================
     # =======================================================================
@@ -82,6 +83,7 @@ class AMQService:
         logging.info("***********************************************")
         logging.info("******  AMQ   Service  Init  Complete *********")
         logging.info("***********************************************")
+        self.future.set_result(True)
         self.loop.run_forever()
 
     def backpressure(self, current_availability: int, maximum: int = -1):
@@ -163,6 +165,48 @@ class AMQService:
             return True
         return False
 
+    async def data_message_generator(self, route: AMQRoute) -> AsyncIterator[DataMessage]:
+        """
+        An async iterator that blocks until messages arrive, processes them,
+        and continues in a loop.
+
+        :param queue_name: The name of the queue to consume messages from
+        :yield: The processed DataResponse for each message
+        """
+        # Get a channel from the connection
+        channel = self.rabbit_mq_client.channel
+        queue_args = (
+            self.rabbit_mq_client.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
+        )
+        queue = await channel.declare_queue(
+            route.queue,
+            durable=True,
+            exclusive=False,
+            auto_delete=False,
+            arguments=queue_args,
+        )
+        await queue.bind(route.exchange, route.key)
+        # Start consuming without a callback
+        async with queue.iterator() as queue_iter:
+            async for message in queue_iter:
+                try:
+                    # Process the message using the same logic as in inbound_data_message_callback_no_thread
+                    data_message: DataMessage = (
+                        await self.amq_data_message_handler.reconstitute_data_message(message)
+                    )
+
+                    success = yield data_message
+                    if not success:
+                        logging_warning(f"Message processing failed for {data_message.id()}.")
+                        # Nack the message without requeuing in case of failure
+                        await message.nack(requeue=False)
+                    else:
+                        await message.ack()
+                except Exception as e:
+                    logging_error(f"Error processing message: {e}")
+                    # Nack the message without requeuing in case of error
+                    await message.nack(requeue=False)
+
     def get_unique_instance_id(self) -> int:
         """
         Provides cluster-wide unique instance ID for this AMQ service instance.

demo.py

@@ -218,7 +218,12 @@ async def get_documents():
                 )
                 otadapter.amq_service.command(
                     "otdemo-2",
-                    "otdemo.otdemo_backend.__global__.print_document_id_static",
+                    "print_document_id_static",
+                    document_id=did + "-static",
+                )
+                otadapter.amq_service.command(
+                    "otdemo-2",
+                    "print_document_id",
                     document_id=did,
                 )
 

otdemo/otdemo.backend.properties

@@ -0,0 +1,21 @@
+# ====================================================================
+# CleverMicro AMQ Adapter settings
+# ====================================================================
+[CleverMicro-AMQ]
+cm.amq-adapter.service-name=otdemo-2
+cm.amq-adapter.generator-id=1
+cm.amq-adapter.route-mapping=otdemo-2:otdemo-exchange:otdemo-queue-2:otdemo-2.#:5:0
+cm.amq-adapter.require-authenticated-user=false
+cm.amq-adapter.callbacks-enabled=False
+
+# CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding
+cm.dispatch.use-dlq=true
+cm.dispatch.amq-host=localhost
+cm.dispatch.amq-port=5672
+cm.dispatch.amq-port-tls=5671
+cm.dispatch.download-dir=/tmp/downloaded_files
+
+cm.backpressure.threshold=1
+cm.backpressure.time-window=150
+cm.backpressure.cpu-overload-duration=300
+cm.backpressure.cpu-idle-duration=300

otdemo/otdemo.properties

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

otdemo/otdemo_backend.py

@@ -1,3 +1,10 @@
+import asyncio
+
+from amqp.config.amq_configuration import AMQConfiguration
+from amqp.model.model import AMQRoute, DataMessage
+from otdemo.otdemo_adapter import OTDemoAmqpAdapter
+
+
 def print_document_id_static(document_id: str) -> str:
     print(f"STATIC Document ID: {document_id}")
     return "STATIC PRINTED!"
@@ -5,21 +12,42 @@ def print_document_id_static(document_id: str) -> str:
 
 class Backend:
 
-    def __init__(self, name: str, description: str):
-        self.name = name
-        self.description = description
-
-    def __repr__(self):
-        return f"Backend(name={self.name}, description={self.description})"
-
-    def __str__(self):
-        return f"{self.name}: {self.description}"
-
-    def __eq__(self, other):
-        if not isinstance(other, Backend):
-            return False
-        return self.name == other.name and self.description == other.description
-
     def print_document_id(self, document_id: str) -> str:
         print(f"CLASS Document ID: {document_id}")
         return "CLASS PRINTED!"
+
+
+async def processing_loop():
+    while (
+        not otadapter.amq_service.future.done()
+    ):  # Wait until AMQ adapter initializes RabbitMQ connection
+        await asyncio.sleep(0.1)
+    # IMPORTANT - in the next line the string arg must match the first token in route-mapping property
+    route: AMQRoute = otadapter.amq_service.router.find_route_by_service("otdemo-2")
+    generator = otadapter.amq_service.data_message_generator(route)  # this is a generator function
+    print(f"Data message generator created: {generator}")
+    message = await generator.asend(None)  # First value must be None to start the generator
+    while True:
+        print(f"Received message: {message}")
+        if message is None:
+            break
+        data_message: DataMessage = message
+        # This is a message processing part. You may use data_message.method() or .domain() or .path()
+        # to access values provided by caller in the send command() call.
+        document_id = data_message._base64body.decode("utf-8")
+        print(
+            f"DATA_MESSAGE Attributes: {data_message.method()} {data_message.domain()} {data_message.path()}"
+        )
+        if data_message.path() == "print_document_id_static":
+            print_document_id_static(document_id)
+        elif data_message.path() == "print_document_id":
+            backend = Backend()
+            backend.print_document_id(document_id)
+        message = await generator.asend(True)
+
+
+if __name__ == "__main__":
+    otadapter = OTDemoAmqpAdapter(AMQConfiguration("otdemo.backend.properties"), [])
+    # Run the processing loop in the AMQ service event loop
+    asyncio.run_coroutine_threadsafe(processing_loop(), otadapter.amq_service.loop)
+    otadapter.thread.join()  # if the code above is not running a loop, need to join the thread to keep program running

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.36"
+version = "0.2.37"
 description = "The Python implementation of the AMQ Adapter for CleverMicro"
 readme = "README.md"
 authors = [