diff --git a/.forgejo/workflows/build-and-publish.yml b/.forgejo/workflows/build-and-publish.yml index dfec56b..1650546 100644 --- a/.forgejo/workflows/build-and-publish.yml +++ b/.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 diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 1db5900..8b958ad 100644 --- a/amqp/service/amq_message_handler.py +++ b/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 @@ -282,23 +283,23 @@ 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), @@ -310,22 +311,20 @@ class DataMessageHandler: 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 + 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", - exchange_name, + rpc_exchange_name, route.component_name, - message_id + message_id, ) - + return future