Messaging

Messaging facilitates communication between applications, with the ArconQ platform handling message queue, security, and connections.

Sending a message

class MessageBus(ABC):
    @abstractmethod
    async def publish_async(
        self,
        message: Message,
        *,
        is_persistent: bool = False,
        routing_id: Optional[str] = None,
    ) -> None:
        pass

In order to send messages IMessageBus interface has been exposed. There are two ways of sending messages:

1. publish_async(message) allows to send message with a custom object message. In this method message object type is used as routingId
2. publish_async(message, routing_id="PythonMessage") allows to overwrite default routingId with a custom message type.
3. publish_async(message, is_persistent=true) allows sending a message with the isPersistent parameter, which, if true, ensures the message is retained for resilience against broker restarts or failures.
4. publish_async(message, routing_id="PythonMessage", is_persistent=true) allows sending a message with a specific routingId and the isPersistent parameter. If isPersistent is true, ensures the message is retained for resilience against broker restarts or failures.

Persistent Messaging

Persistent messaging guarantees the safety of messages during disruptions like outages or restarts of the message broker. By marking a message as persistent, it's ensured that the message will be held until it has been delivered successfully. This feature is vital for applications that depend on the reliable delivery of messages, ensuring that essential communications are preserved even in the face of system interruptions. The storage method for persistent messages safeguards them from being lost, providing a critical layer of security for the integrity of data and the robustness of the system.

Receiving a message

caution

When defining TMessage classes in Sender and Receiver, where one of them uses Python SDK and another .NET SDK remember about Namespace alignment for Sender and Receiver rule. For communication between Python apps it's not necessary.

In order to receive a message you need to create a MessageHandler object and use it to decorate a function with your handling logic using @handler.handle.

from arconq_sdk import Dependency, Logger, MessageHandler
from containers.container import Container
from contracts.example_message import ExampleMessage

handler = MessageHandler()

@handler.handle("/example-handler", message_type=ExampleMessage)
async def example_message_handler(
    message: ExampleMessage,
    logger: Logger = Dependency(Container.logger),
):
    logger.info("Handler received a message of content: {Message}", Message=message)

The next step is registering your message handler.

@arconq_app
def main(
    container: Container,
    app_builder: AppBuilder = Dependency(Container.app_builder),
) -> None:
    app_builder.configure_arconq(container)
    
    app_builder.add_message_handler(container, example_message_handler, example_message_handler.handler)

    app = app_builder.build_app()
    app.run()

Message Retries

The retry mechanism allows for automatic retries when sending messages using the ArconQ SDK. This feature can be enabled by adding the retry configuration to the message handler registration. To enable the retry mechanism, use the WithRetries extension method when configuring the message handler. This method specifies the number of retries that should be attempted if sending the message fails.

The retry mechanism in utilizes an exponential backoff strategy when retrying failed message sending attempts. This strategy introduces a delay between each retry, increasing the delay duration exponentially with each subsequent retry. The exponential backoff strategy helps to prevent overwhelming the system with repeated failed attempts and provides a more efficient and scalable approach to retrying.

@handler.handle("/example-handler", message_type=ExampleMessage, retries_count=5)

In this example, the message will be sent up to 6 times in total (1 initial attempt + 5 retries).

info

If the message is processed successfully before reaching the maximum retries, no additional retries will be performed.
If all of the retries fail, the message will go to the DeadLetterQueue and there will be no other attempts to deliver it.

To trigger a retry, an error should be thrown inside the message handler. This error will be caught by the retry mechanism, and the message will be resent according to the configured retry settings.

@handler.handle("/example-handler", message_type=ExampleMessage)
async def example_message_handler(
    message: ExampleMessage
):
    try:
        # Message handling logic 
    except:
        # Throw an error to trigger retry
        raise Exception("Failed to send message")

Retry Configuration

The retry configuration can be customized according to your application's requirements. The WithRetries method accepts an integer parameter that represents the maximum number of retries.

• 0 retries means the message will be sent only once.
• 1 retry means the message will be sent again if the initial attempt fails.
• N retries means the message will be sent up to N+1 times, including the initial attempt.

Examples

Using class name as routingId

1. Sending message

from arconq_sdk.contracts.message import Message


class ExampleMessage(Message):
    Id: int
    Description: str
    Number: int

from arconq_sdk import Logger, MessageBus
from contracts.example_message import ExampleMessage


class MessagingService:
    def __init__(self, logger: Logger, message_bus: MessageBus) -> None:
        self.logger = logger
        self.message_bus = message_bus

    async def send_message(self) -> None:
        message = ExampleMessage(
            Id=1,
            Description="message sent by a Python app",
            Number=22,
        )
        self.logger.info("Sending a message: {Message}", Message=message)
        await self.message_bus.publish_async(message)

2. Receiving message

Message Handler:

from arconq_sdk import Dependency, Logger, MessageHandler
from containers.container import Container
from contracts.example_message import ExampleMessage

handler = MessageHandler()

@handler.handle("/example-handler", message_type=ExampleMessage)
async def example_message_handler(
    message: ExampleMessage,
    logger: Logger = Dependency(Container.logger),
):
    logger.info("Handler received a message of content: {Message}", Message=message)

info

In this scenario, the message handler handles only messages without a custom routingId, with the same class name.

Using Custom routingId

1. Sending message with custom routingId

from arconq_sdk.contracts.message import Message


class ExampleMessage(Message):
    Id: int
    Description: str
    Number: int

from arconq_sdk import Logger, MessageBus
from contracts.example_message import ExampleMessage


class MessagingService:
    def __init__(self, logger: Logger, message_bus: MessageBus) -> None:
        self.logger = logger
        self.message_bus = message_bus

    async def send_message(self) -> None:
        message = ExampleMessage(
            Id=1,
            Description="message sent by a Python app",
            Number=22,
        )
        self.logger.info("Sending a message: {Message}", Message=message)
        await self.message_bus.publish_async(message, routing_id="MessageType001")

2. Receiving message with custom routingId

Message Handler:

from arconq_sdk import Dependency, Logger, MessageHandler
from containers.container import Container
from contracts.example_message import ExampleMessage

handler = MessageHandler()

@handler.handle("/example-handler", message_type=ExampleMessage, routing_id="MessageType001")
async def example_message_handler(
    message: ExampleMessage,
    logger: Logger = Dependency(Container.logger),
):
    logger.info("Handler received a message of content: {Message}", Message=message)

info

Please note that you can use different class than MyMessage in the message handler. It's important that the fields in class are the same as the fields from received message.
It means that @handler.handle("/example-handler", message_type=DifferentMessage, routing_id="MessageType001") also handles messages with class name MyMessage routed via MessageType001

Using Persistent messaging

1. Sending a persistent message

from arconq_sdk.contracts.message import Message


class ExampleMessage(Message):
    Id: int
    Description: str
    Number: int

from arconq_sdk import Logger, MessageBus
from contracts.example_message import ExampleMessage


class MessagingService:
    def __init__(self, logger: Logger, message_bus: MessageBus) -> None:
        self.logger = logger
        self.message_bus = message_bus

    async def send_message(self) -> None:
        message = ExampleMessage(
            Id=1,
            Description="message sent by a Python app",
            Number=22,
        )
        self.logger.info("Sending a persistent message: {Message}", Message=message)
        await self.message_bus.publish_async(message, is_persistent=True)