feat/#61 - add one way communication option

Co-authored-by: aider (openrouter/anthropic/claude-3.7-sonnet) <aider@aider.chat>

.forgejo/workflows/build-and-publish.yml

@@ -22,7 +22,7 @@ env:
 
 jobs:
   build-and-push:
-    runs-on: ubuntu-latest
+    runs-on: general
     services:
       dind:
         image: docker:dind

amqp/rabbitmq/user_management_sample.py

@@ -1,11 +1,11 @@
 import asyncio
 import logging
-from typing import Dict, Any
+from typing import Any, Dict
 
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.rabbitmq.user_management_service_client import (
     UserManagementServiceClient,
-    UserManagementServiceException
+    UserManagementServiceException,
 )
 
 
@@ -14,39 +14,39 @@ async def get_user_info_from_token(
 ) -> Dict[str, Any]:
     """
     Get user information from a JWT token.
-    
+
     Args:
         client: User Management Service client
         token: JWT token
-        
+
     Returns:
         User information
-        
+
     Raises:
         UserManagementServiceException: If an error occurs
     """
     try:
         # Validate the token
         token_info = await client.validate_token(token)
-        
+
         # Extract user ID from token info
         user_id = None
         for item in token_info:
             if isinstance(item, dict) and "sub" in item:
                 user_id = item["sub"]
                 break
-        
+
         if not user_id:
             raise UserManagementServiceException(
                 "cleverthis.clevermicro.auth.invalid_token",
                 "Token does not contain user ID (sub claim)",
-                {}
+                {},
             )
-        
+
         # Query user information
         user_info = await client.query_user_by_id(user_id)
         return user_info
-        
+
     except UserManagementServiceException as e:
         logging.error(f"User Management Service error: {e.exception_type}: {e.message}")
         raise
@@ -57,19 +57,19 @@ async def get_user_info_from_token(
 
 async def main():
     # Initialize configuration
-    config = AMQConfiguration()
-    
+    config = AMQConfiguration("")
+
     # Create client
     client = UserManagementServiceClient(config)
-    
+
     try:
         # Sample JWT token (replace with actual token)
         token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c"
-        
+
         # Get user information
         user_info = await get_user_info_from_token(client, token)
         print(f"User information: {user_info}")
-        
+
     except UserManagementServiceException as e:
         print(f"Error: {e.exception_type}: {e.message}")
         if e.additional_info:

amqp/service/amq_message_handler.py

@@ -10,6 +10,7 @@ from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
 from opentelemetry import trace
 from opentelemetry.trace import Tracer, TraceState
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+from pika.exchange_type import ExchangeType
 
 from amqp.adapter.backpressure_handler import BackpressureHandler
 from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
@@ -160,6 +161,12 @@ class DataMessageHandler:
     async def reconstitute_data_message(
         self, message: AbstractIncomingMessage
     ) -> DataMessage | None:
+        """
+        DataMessage may be serialized as string in separate queue (addressing == 1) or be a MultiPart message with
+        files streamed over individual queues, one per file.
+        This function uses 'addressing' mode from message headers to determine how to deserialize (reconstitute)
+        the original message.
+        """
         amq_message: DataMessage | None = None
         delivery_tag = message.delivery_tag
         addressing: int = message.headers.get("clevermicro.addressing", 0)
@@ -246,8 +253,8 @@ class DataMessageHandler:
 
     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.
+        The handler for reply for DataMessage that we sent out earlier -
+        the code matches the received response with the original request and passes the response to the caller.
         :param message:
         :return:
         """
@@ -276,3 +283,85 @@ class DataMessageHandler:
             future.set_result(response.body())
 
         await message.ack(multiple=False)
+
+    async def send_rpc(self, route, message: DataMessage) -> Future:
+        """
+        Sends an RPC message to the specified route and returns a Future that will be completed
+        when the response is received.
+
+        :param route: AMQRoute object containing exchange and routing information
+        :param message: DataMessage to be sent
+        :return: Future that will be completed with the response payload
+        """
+        # Create a Future to be completed when the response is received
+        future = asyncio.Future()
+
+        # Store the future in the outstanding dictionary using the message ID as the key
+        message_id = message.id().as_string()
+        self.outstanding[message_id] = future
+
+        # Create a Pika message with the correlation ID set to the message ID
+        pika_message = Message(
+            body=DataMessageFactory.serialize(message),
+            correlation_id=message_id,
+            reply_to=self.rabbit_mq_client.router.get_reply_to_queue_name(),
+            content_type=(
+                message.content_type()[0]
+                if isinstance(message.content_type(), list)
+                else message.content_type()
+            ),
+        )
+
+        # Get the exchange from the route
+        rpc_exchange_name = route.exchange
+        exchange: AbstractRobustExchange = await self.rabbit_mq_client.channel.declare_exchange(
+            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)
+
+        logging_debug(
+            "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s",
+            rpc_exchange_name,
+            route.component_name,
+            message_id,
+        )
+
+        return future
+
+    async def send_command(self, route, message: DataMessage) -> bool:
+        """
+        Sends an RPC message to the specified route and returns a Future that will be completed
+        when the response is received.
+
+        :param route: AMQRoute object containing exchange and routing information
+        :param message: DataMessage to be sent
+        :return: Future that will be completed with the response payload
+        """
+        # Create a Pika message with the correlation ID set to the message ID
+        pika_message = Message(
+            body=DataMessageFactory.serialize(message),
+            correlation_id=message.id().as_string(),
+            content_type=(
+                message.content_type()[0]
+                if isinstance(message.content_type(), list)
+                else message.content_type()
+            ),
+        )
+
+        # Get the exchange from the route
+        rpc_exchange_name = route.exchange
+        exchange: AbstractRobustExchange = await self.rabbit_mq_client.channel.declare_exchange(
+            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)
+
+        logging_debug(
+            "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s",
+            rpc_exchange_name,
+            route.component_name,
+            message.id().as_string(),
+        )
+
+        return True