diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 34f593c..99fc55a 100644 --- a/amqp/service/amq_message_handler.py +++ b/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: diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index d7b82d0..a357abe 100644 --- a/amqp/service/amq_service.py +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 87be1c7..28818e0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.34" +version = "0.2.35" description = "The Python implementation of the AMQ Adapter for CleverMicro" readme = "README.md" authors = [ diff --git a/tests/adapter/test_rpc.py b/tests/adapter/test_rpc.py index d21d0c2..d7b3d2a 100644 --- a/tests/adapter/test_rpc.py +++ b/tests/adapter/test_rpc.py @@ -1,6 +1,7 @@ import importlib import sys import unittest + from amqp.service.amq_message_handler import invoke_command @@ -43,7 +44,9 @@ class MyService: self.my_module = importlib.util.module_from_spec(spec) exec(dummy_module_content, self.my_module.__dict__) - sys.modules["my_package"] = self.my_module # Add to sys.modules for importlib to find + sys.modules["my_package.my_module"] = ( + self.my_module + ) # Add to sys.modules for importlib to find async def asyncTearDown(self): """Clean up after tests.""" @@ -57,9 +60,9 @@ class MyService: package_name="my_package.my_module", class_name="__global__", function_name="greet_sync", - kwargs={"name": "Alice"} + kwargs={"name": "Alice"}, ) - self.assertEqual(result, "Hello, Alice (sync global)!") + self.assertEqual(result.body().decode(), "Hello, Alice (sync global)!") async def test_invoke_global_async_function(self): """Test invoking a global asynchronous function.""" @@ -67,9 +70,9 @@ class MyService: package_name="my_package.my_module", class_name="__global__", function_name="greet_async", - kwargs={"name": "Bob"} + kwargs={"name": "Bob"}, ) - self.assertEqual(result, "Hello, Bob (async global)!") + self.assertEqual(result.body().decode(), "Hello, Bob (async global)!") async def test_invoke_class_sync_method(self): """Test invoking a class synchronous method.""" @@ -77,11 +80,11 @@ class MyService: package_name="my_package.my_module", class_name="MyService", function_name="process_data_sync", - kwargs={"data": "report", "count": 3} + kwargs={"data": "report", "count": 3}, ) self.assertEqual( - result, - "Service 'MyAwesomeService' processed 'report' 3 times (sync method)." + result.body().decode(), + "Service 'MyAwesomeService' processed 'report' 3 times (sync method).", ) async def test_invoke_class_async_method(self): @@ -90,11 +93,11 @@ class MyService: package_name="my_package.my_module", class_name="MyService", function_name="process_data_async", - kwargs={"data": "metrics", "delay": 0.01} + kwargs={"data": "metrics", "delay": 0.01}, ) self.assertEqual( - result, - "Service 'MyAwesomeService' processed 'metrics' after 0.01s (async method)." + result.body().decode(), + "Service 'MyAwesomeService' processed 'metrics' after 0.01s (async method).", ) async def test_package_not_found(self): @@ -104,7 +107,7 @@ class MyService: package_name="non_existent_package", class_name="__global__", function_name="some_func", - kwargs={} + kwargs={}, ) async def test_class_not_found(self): @@ -114,7 +117,7 @@ class MyService: package_name="my_package.my_module", class_name="NonExistentClass", function_name="some_method", - kwargs={} + kwargs={}, ) async def test_method_not_found(self): @@ -124,7 +127,7 @@ class MyService: package_name="my_package.my_module", class_name="MyService", function_name="non_existent_method", - kwargs={} + kwargs={}, ) async def test_private_method_access(self): @@ -133,9 +136,10 @@ class MyService: package_name="my_package.my_module", class_name="MyService", function_name="_private_method", - kwargs={} + kwargs={}, ) - self.assertEqual(result, "This is a private method.") + self.assertEqual(result.body().decode(), "This is a private method.") + if __name__ == "__main__": unittest.main()