feat/#63 - fix the send_command and add unit tests

amqp/service/amq_message_handler.py

@@ -1,9 +1,11 @@
 import asyncio
+import importlib
+import inspect
 import os
 import threading
 import time
 from asyncio import AbstractEventLoop, Future
-from typing import Dict
+from typing import Any, Dict
 
 from aio_pika import Message
 from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
@@ -18,7 +20,7 @@ from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.data_response_factory import DataResponseFactory
 from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
 from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
-from amqp.adapter.serializer import map_as_string
+from amqp.adapter.serializer import map_as_string, parse_map_string
 from amqp.adapter.service_message_factory import ServiceMessageFactory
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import (
@@ -46,6 +48,75 @@ def set_text(carrier, key, value):
     carrier[key] = value
 
 
+async def invoke_command(
+    package_name: str, class_name: str, function_name: str, msg_id: str = "", kwargs=None
+) -> Any:
+    if kwargs is None:
+        kwargs = {}
+    try:
+        # 1. Dynamically import the package
+        logging_info(f"Attempting to import package: {package_name}")
+        package = importlib.import_module(package_name)
+        logging_info(f"Package '{package_name}' imported successfully.")
+
+        target_callable = None
+
+        if class_name == "__global__":
+            # 2. Handle global function invocation
+            logging_info(f"Attempting to find global function: {function_name} in {package_name}")
+            target_callable = getattr(package, function_name)
+            logging_info(f"Global function '{function_name}' found.")
+        else:
+            # 3. Handle class method invocation
+            logging_info(f"Attempting to find class: {class_name} in {package_name}")
+            target_class = getattr(package, class_name)
+            logging_info(f"Class '{class_name}' found. Instantiating...")
+
+            # Instantiate the class. Assuming a default constructor or one that doesn't
+            # require arguments from kwargs for initialization.
+            # If the class __init__ needs specific arguments, this part might need adjustment
+            # based on how those arguments are provided.
+            instance = target_class()
+            logging_info(f"Instance of '{class_name}' created.")
+
+            logging_info(f"Attempting to find method: {function_name} in class {class_name}")
+            target_callable = getattr(instance, function_name)
+            logging_info(f"Method '{function_name}' found.")
+
+        # 4. Invoke the function/method
+        logging_info(f"Invoking '{function_name}' with arguments: {kwargs}")
+        if inspect.iscoroutinefunction(target_callable):
+            result = await target_callable(**kwargs)
+            logging_info(f"Asynchronous function '{function_name}' invoked. Result: {result}")
+        else:
+            result = target_callable(**kwargs)
+            logging_info(f"Synchronous function '{function_name}' invoked. Result: {result}")
+
+        response: DataResponse = DataResponseFactory.of(
+            id=msg_id,
+            response_code=200,
+            content_type="text/plain",
+            body=result.encode("utf-8"),
+            trace_info=collect_trace_info(),
+            error=None,
+            error_cause=None,
+        )
+        response._magic = DataMessageMagicByte.RPC_REQUEST.value
+        return response
+
+    except ModuleNotFoundError as e:
+        logging_error(f"Error: Package '{package_name}' not found. {e}")
+        raise
+    except AttributeError as e:
+        logging_error(
+            f"Error: Class '{class_name}' or function/method '{function_name}' not found in '{package_name}'. {e}"
+        )
+        raise
+    except Exception as e:
+        logging_error(f"An unexpected error occurred during invocation: {e}")
+        raise
+
+
 class DataMessageHandler:
     def __init__(
         self,
@@ -98,6 +169,8 @@ class DataMessageHandler:
             self.ensure_trace_info(amq_message)
             if amq_message.magic() == DataMessageMagicByte.HTTP_REQUEST.value:
                 await self.run_http_request_magic(message, amq_message, self.loop)
+            elif amq_message.magic() == DataMessageMagicByte.RPC_REQUEST.value:
+                await self.run_rpc_request(message, amq_message, self.loop)
             else:
                 logging_error(
                     "Unknown MAGIC value: [%s], don't know how to interpret the message",
@@ -158,6 +231,85 @@ class DataMessageHandler:
 
         self.backpressure_handler.decrease_current_load()
 
+    async def run_rpc_request(
+        self,
+        message: AbstractIncomingMessage,
+        amq_message: DataMessage,
+        loop: AbstractEventLoop,
+    ) -> None:
+        """
+        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
+        """
+        # Increment the counter for parallel executions and check resource availability
+        self.backpressure_handler.increase_current_load()
+        await self.backpressure_handler.check_overload_condition()
+        logging_info(
+            f"PARALLEL: Current Capacity [{self.backpressure_handler.current_availability}] of [{self.backpressure_handler.max_availability}]"
+        )
+        if message.reply_to:
+            self.reply_to = message.reply_to
+
+        try:
+            _a_message = AMQMessage(amq_message, self.rabbit_mq_client.connection)
+            _response: DataResponse = await self.service_adapter.on_rpc(_a_message)
+            logging_debug(f"Result in thread: {_response}")
+            try:
+                await self.send_reply(message.reply_to, _response)
+            except Exception as e:
+                logging_error(f"Processing message: {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"Deleting file {filename}: {e}")
+        except Exception as e:
+            logging_error(f"Request handler: {e}")
+
+        self.backpressure_handler.decrease_current_load()
+
+    async def on_rpc(self, a_message: AMQMessage) -> DataResponse:
+        """
+        Dynamically invokes a function or a class method from a specified package.
+
+        Args:
+            a_message (AMQMessage): The message containing the package, class, and function details.
+        Args encoded in the AMQMessage:
+            package_name (str): The name of the Python package to import.
+            class_name (str): The name of the class within the package.
+                              Use '__global__' if the function_name refers to a global function
+                              in the package.
+            function_name (str): The name of the function or method to invoke.
+            kwargs (Dict[str, Any]): A dictionary of keyword arguments to pass to the function/method (in the body).
+
+        Returns:
+            Any: The return value of the invoked function or method.
+
+        Raises:
+            ModuleNotFoundError: If the specified package cannot be imported.
+            AttributeError: If the specified class or function/method does not exist.
+            Exception: For any other errors during invocation.
+        """
+        package_name: str = a_message.domain()
+        class_name: str = a_message.path()
+        function_name: str = a_message.method()
+        kwargs: Dict[str, Any] = parse_map_string(a_message.body().decode("utf-8"))
+
+        return await invoke_command(
+            package_name, class_name, function_name, a_message.id().as_string(), kwargs
+        )
+
     async def reconstitute_data_message(
         self, message: AbstractIncomingMessage
     ) -> DataMessage | None:

amqp/service/amq_service.py

@@ -128,7 +128,10 @@ class AMQService:
                 swarm_id=self.amq_configuration.amq_adapter.swarm_task_id,
                 **kwargs,
             )
-            result: Future = self.amq_data_message_handler.send_rpc(route, message)
+            result: Future = asyncio.run_coroutine_threadsafe(
+                self.amq_data_message_handler.send_rpc(route, message),
+                loop=self.backpressure_handler.loop,
+            )
             self.set_outstanding_future(message, result)
             return result
 
@@ -138,6 +141,28 @@ class AMQService:
         )
         return result
 
+    def command(self, service_name: str, fq_method_name: str, **kwargs) -> bool:
+        """
+        Send a unidirectional Command message to the AMQ service.
+        :param service_name: The service name to receive the RPC message.
+        :param fq_method_name: The fully qualified method name to execute the RPC message.
+        :param kwargs: Additional keyword arguments to be passed to the RPC method.
+        :return: A Future that resolves when the RPC response is received.
+        """
+        route: Optional[AMQRoute] = self.router.find_route_by_service(service_name)
+        if route is not None:
+            message: DataMessage = self.data_message_factory.create_rpc_message(
+                fq_method_name=fq_method_name,
+                swarm_id=self.amq_configuration.amq_adapter.swarm_task_id,
+                **kwargs,
+            )
+            asyncio.run_coroutine_threadsafe(
+                self.amq_data_message_handler.send_command(route, message),
+                loop=self.backpressure_handler.loop,
+            )
+            return True
+        return False
+
     def get_unique_instance_id(self) -> int:
         """
         Provides cluster-wide unique instance ID for this AMQ service instance.