diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index bcc7030..1db5900 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -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