From f02823e18dbdd173dafeed24219c0e1150deb192 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Fri, 18 Jul 2025 07:54:55 +0100 Subject: [PATCH 1/4] feat/#61 - add client code for User Managemengt Service --- amqp/rabbitmq/user_management_sample.py | 30 ++++++++++++------------- amqp/service/amq_message_handler.py | 10 +++++++-- 2 files changed, 23 insertions(+), 17 deletions(-) diff --git a/amqp/rabbitmq/user_management_sample.py b/amqp/rabbitmq/user_management_sample.py index b6a2c72..3f3d4ba 100644 --- a/amqp/rabbitmq/user_management_sample.py +++ b/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: diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 8dea0e4..bcc7030 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -160,6 +160,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 +252,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: """ From cf1804d23f3c3ca66b1f17b090ba466a73f24f1f Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Fri, 18 Jul 2025 10:29:08 +0100 Subject: [PATCH 2/4] feat: implement send_rpc method in DataMessageHandler class Co-authored-by: aider (openrouter/anthropic/claude-3.7-sonnet) --- amqp/service/amq_message_handler.py | 47 +++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) 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 From 3ee62b043700be9d86c482eb4cd2975526bde594 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Mon, 21 Jul 2025 10:49:28 +0100 Subject: [PATCH 3/4] feat/#58 - fix unit test and deployment pipeline --- .forgejo/workflows/build-and-publish.yml | 2 +- amqp/service/amq_message_handler.py | 31 ++++++++++++------------ 2 files changed, 16 insertions(+), 17 deletions(-) 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 From b41ca04801138b5d9cdbc1c18626736bf3195ac3 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Wed, 23 Jul 2025 18:33:34 +0100 Subject: [PATCH 4/4] feat/#61 - add one way communication option --- amqp/service/amq_message_handler.py | 39 ++++++++++++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 8b958ad..34f593c 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -297,7 +297,7 @@ class DataMessageHandler: future = asyncio.Future() # Store the future in the outstanding dictionary using the message ID as the key - message_id = message.id() + message_id = message.id().as_string() self.outstanding[message_id] = future # Create a Pika message with the correlation ID set to the message ID @@ -328,3 +328,40 @@ class DataMessageHandler: ) 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