Messaging

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

Sending a message

 public interface IMessageBus
    {
        Task PublishAsync<TMessage>(TMessage message);

        Task PublishAsync<TMessage>(string routingId, TMessage message);

        Task PublishAsync<TMessage>(TMessage message, bool isPersistent);

        Task PublishAsync<TMessage>(string routingId, TMessage message, bool isPersistent);
    }

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

1. Task PublishAsync <TMessage>(TMessage message) allows to send message with a custom object <TMessage>. In this method <TMessage> object type is used as routingId
2. Task PublishAsync<TMessage>(string routingId, TMessage message) allows to overwrite default routingId with a custom message type.
3. Task PublishAsync<TMessage>(TMessage message, bool isPersistent) allows sending a message with the isPersistent parameter, which, if true, ensures the message is retained for resilience against broker restarts or failures.
4. Task PublishAsync<TMessage>(string routingId, TMessage message, bool isPersistent) 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 remember about Namespace alignment for Sender and Receiver rule.

 public interface IMessageHandler<TMessage>
    {
        Task HandleAsync(TMessage message);
    }

In order to receive a message it's necessary to implement IMessageHandler interface. It handles message of custom type TMessage
The next step is registering your message handler.

await ArconQApp.StartAsync(args,
    services =>
    {
        // Registers MyMessageHandler which handles all messages routed via MyMessage.
        // It ignores messages routed via custom routingId
        services.AddMessageHandler<MyMessageHandler, MyMessage>();

        // Registers MySecondMessageHandler which handles all messages routed via MessageTypeX001.
        // It handles messages of all types.
        services.AddMessageHandler<MySecondMessageHandler, MySecondMessage>().WithRoutingId("MessageTypeX001");
    });

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.

builder.Services.AddMessageHandler<MyMessageHandler, MyMessage>().WithRetries(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.

public class MyMessageHandler : IMessageHandler<MyMessage>
{
    public async Task HandleAsync(MyMessage message)
    {
        try
        {
            // Message handling logic
        }
        catch (Exception ex)
        {
            // Throw an error to trigger retry
            throw new Exception("Failed to send message", ex);
        }
    }
}

Namespace alignment for Sender and Receiver

If a RoutingId is not specified, the class names for the Receiver and Sender applications must be identical for message routing. Additionally, both classes must reside in the same namespace.

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

    public class ExampleMessageSender : IExampleMessageSender
    {
        private readonly IMessageBus _messageBus;
        private readonly ILogger<ExampleMessageSender> _logger;
        public ExampleMessageSender(IMessageBus messageBus, ILogger<ExampleMessageSender> logger)
        {
            _messageBus = messageBus;
            _logger = logger;
        }
        public async void SendMessage()
        {
            await _messageBus.PublishAsync(new MyMessage() {Id = 1, Name = "Test"});
        }
    }

    public class MyMessage
    {
        public int Id { get; set; }
        public string Name { get; set; }
    }

2. Receiving message

Message Handler:

internal class MyMessageHandler : IMessageHandler<MyMessage>
{
    private readonly ILogger<MyMessageHandler> _logger;

    public MyMessageHandler(ILogger<MyMessageHandler> logger)
    {
        _logger = logger;
    }

    public Task HandleAsync(MyMessage message)
    {
        _logger.LogInformation($"I have a message with id {message.Id}, name {message.Name}");
        return Task.CompletedTask;
    }
}

Message Handler Registration in program.cs:

await ArconQApp.StartAsync(args,
    services =>
        services.AddMessageHandler<MyMessageHandler, MyMessage>();
    });

info

In this scenario, the message handler handles only messages without a custom routingId, with the same class name, residing in the same namespace as in the received message

Using Custom routingId

1. Sending message with custom routingId

    public class ExampleMessageSender : IExampleMessageSender
    {
        private readonly IMessageBus _messageBus;
        private readonly ILogger<ExampleMessageSender> _logger;
        public ExampleMessageSender(IMessageBus messageBus, ILogger<ExampleMessageSender> logger)
        {
            _messageBus = messageBus;
            _logger = logger;
        }
        public async void SendMessage()
        {
            await _messageBus.PublishAsync("MessageType001", new MyMessage() {Id = 1, Name = "Test"});
        }
    }

    public class MyMessage
    {
        public int Id { get; set; }
        public string Name { get; set; }
    }

2. Receiving message with custom routingId

Message Handler:

internal class MyMessageHandler : IMessageHandler<MyMessage>
{
    private readonly ILogger<MyMessageHandler> _logger;

    public MyMessageHandler(ILogger<MyMessageHandler> logger)
    {
        _logger = logger;
    }

    public Task HandleAsync(MyMessage message)
    {
        _logger.LogInformation($"I have a message with id {message.Id}, name {message.Name}");
        return Task.CompletedTask;
    }
}

Message Handler Registration in Program.cs:

await ArconQApp.StartAsync(args,
    services =>
        services.AddMessageHandler<MyMessageHandler, MyMessage>().WithRoutingId("MessageType001");
    });

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 services.AddMessageHandler<MyMessageHandler, MyDifferentMessage>().WithRoutingId("MessageType001") also handles messages with class name MyMessage routed via MessageType001

Using Persistent messaging

1. Sending a persistent message

    public class PersistentMessageSender : IExampleMessageSender
    {
        private readonly IMessageBus _messageBus;
        private readonly ILogger<PersistentMessageSender> _logger;

        public PersistentMessageSender(IMessageBus messageBus, ILogger<PersistentMessageSender> logger)
        {
            _messageBus = messageBus;
            _logger = logger;
        }

        public async void SendPersistentMessageAsync()
        {
            var message = new PersistentMessage { Id = 1, Content = "Info" };
            await _messageBus.PublishAsync(message, true);
            _logger.LogInformation("Sent a persistent message.");
        }
    }

    public class PersistentMessage
    {
        public int Id { get; set; }
        public string Content { get; set; }
    }