feat: implement send_rpc method in DataMessageHandler class

Co-authored-by: aider (openrouter/anthropic/claude-3.7-sonnet) <aider@aider.chat>
2025-07-18 10:29:08 +01:00
parent 04f1483c06
commit cf1804d23f
1 changed files with 47 additions and 0 deletions
@@ -282,3 +282,50 @@ 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()
+        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
+        exchange_name = route.exchange
+        exchange = await self.rabbit_mq_client.connection.get_exchange(exchange_name)
+        
+        # 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",
+            exchange_name,
+            route.component_name,
+            message_id
+        )
+        
+        return future