Storage | Messaging | Reflection
The Namotion.Messaging .NET libraries provide abstractions and implementations for message brokers, event queues and data ingestion services.
By programming against a messaging abstraction you enable the following scenarios:
To listen for messages, create a new message receiver for a specific implementation and call the ListenWithRetryAsync()
method (this will newer return):
IMessageReceiver receiver = ServiceBusMessageReceiver
.Create("MyConnectionString", "myqueue");
await receiver.ListenAsync(async (messages, ct) =>
{
...
}, CancellationToken.None);
In another process or thread you can then publish messages to this listener:
IMessagePublisher publisher = ServiceBusMessagePublisher
.Create(configuration["ServiceBusConnectionString"], "myqueue");
await publisher.PublishAsync(new Message(content: new byte[] { 1, 2, 3 }));
However, you should host the listener with a .NET generic host which provides dependency injection, configuration and logging. To use the IMessageReceiver
in a simple command line application (.NET Generic Host), implement a new BackgroundService
and start message processing in ExecuteAsync
:
public class MyBackgroundService : BackgroundService
{
private readonly IMessageReceiver _messageReceiver;
private readonly ILogger _logger;
public MyBackgroundService(IMessageReceiver messageReceiver, ILogger logger)
{
_messageReceiver = messageReceiver;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await _messageReceiver.ListenWithRetryAsync(async (messages, ct) =>
{
foreach (var message in messages)
{
try
{
// TODO: Process message
await _messageReceiver.ConfirmAsync(message, ct);
}
catch (Exception e)
{
_logger.LogError(e, $"Error while processing {nameof(MyMessage)} message.");
await _messageReceiver.RejectAsync(message, ct);
}
}
}, stoppingToken);
}
}
In your program's Main
method, create a new HostBuilder
and add the background service as a hosted service:
public static async Task Main(string[] args)
{
var host = new HostBuilder()
.ConfigureServices(services =>
{
var receiver = ServiceBusMessageReceiver.Create("MyConnectionString", "myqueue");
services.AddSingleton<IMessageReceiver>(receiver);
services.AddHostedService<MyBackgroundService>();
})
.Build();
await host.RunAsync();
}
Behavior extensions, for example custom dead letter queues or large message handling, is achieved with interceptors which wrap publisher and receiver methods with custom code. These interceptors are added with the With*
extension methods. Custom interceptors can be implemented with the MessagePublisher<T>
and MessageReceiver<T>
classes.
Contains the messaging abstractions, mainly interfaces with a very small footprint and extremely stable contracts:
PublishAsync(messages, cancellationToken)
: Sends a batch of messages to the queue.GetMessageCountAsync(cancellationToken)
: Gets the count of messages waiting to be processed.ListenAsync(handleMessages, cancellationToken)
: (Fails when connection cannot be established)ListenWithRetryAsync(handleMessages, cancellationToken)
: Starts listening and processing messages with the handleMessages
function until the cancellationToken
signals a cancellation.KeepAliveAsync(messages, timeToLive, cancellationToken)
: Extends the message lock timeout on the given messages.ConfirmAsync(messages, cancellationToken)
: Confirms the processing of messages and removes them from the queue.RejectAsync(messages, cancellationToken)
: Rejects messages and requeues them for later reprocessing.DeadLetterAsync(messages, reason, errorDescription, cancellationToken)
: Removes the messages and moves them to the dead letter queue.The idea behind the generic interfaces is to allow multiple instance registrations, read Dependency Injection in .NET: A way to work around missing named registrations for more information.
Contains common helper methods and base implementations of the abstractions:
Extension methods to enhance or modify instances:
IMessagePublisher
/IMessageReceiver
to IMessagePublisher<T>
/IMessageReceiver<T>
.DeadLetterAsync()
will confirm the message and publish it to the specified messagePublisher
.Other extension methods:
Extension methods to enhance or modify instances:
Dependencies:
Provides extension methods on IMessagePublisher<T>
and IMessageReceiver<T>
to enable JSON serialization for messages:
Message<T>.Object
property. If the content could not be deserialized then Object
is null
.Send a JSON encoded message:
var publisher = ServiceBusMessagePublisher
.Create("MyConnectionString", "myqueue")
.AsPublisher<OrderCreatedMessage>();
await publisher.PublishAsJsonAsync(new OrderCreatedMessage { ... });
Receive JSON encoded messages:
var receiver = ServiceBusMessageReceiver
.Create("MyConnectionString", "myqueue")
.AsReceiver<OrderCreatedMessage>();
await receiver.ListenAndDeserializeJsonAsync(async (messages, ct) =>
{
foreach (OrderCreatedMessage message in messages.Select(m => m.Object))
{
...
}
await receiver.ConfirmAsync(messages, ct);
});
The following packages should only be used in the head project, i.e. directly in your application bootstrapping project where the dependency injection container is initialized.
Azure Service Bus |
Azure Event Hub |
Azure Storage Queue |
RabbitMQ | Amazon SQS | InMemory | |
---|---|---|---|---|---|---|
PublishAsync | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: |
ListenAsync | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: |
GetMessageCountAsync | :x: | :x: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: |
KeepAliveAsync | :heavy_check_mark: | :heavy_minus_sign: (1.) | :heavy_check_mark: | :x: | :heavy_check_mark: | :heavy_minus_sign: |
ConfirmAsync | :heavy_check_mark: | :heavy_minus_sign: (1.) | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_minus_sign: |
RejectAsync | :heavy_check_mark: | :heavy_minus_sign: (1.) | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: |
DeadLetterAsync | :heavy_check_mark: | :x: (2.) | :x: (2.) | :x: (2.) | :x: (2.) | :heavy_check_mark: |
User properties | :heavy_check_mark: | :heavy_check_mark: | :x: (3.) | :heavy_check_mark: | :heavy_check_mark: | :heavy_check_mark: |
1) Because Event Hub is stream based and not transactional, these method calls are just ignored.
2) Use receiver.WithDeadLettering(publisher)
to enable dead letter support.
3) Use receiver.WithPropertiesInContent()
to enable user properties support (not implemented yet).
:heavy_minus_sign: = Noop/Ignored
Implementations:
Behavior:
handleMessages
throws an exception, then the messages are abandoned and later reprocessed until they are moved to the dead letter queue.Dependencies:
Implementations:
Behavior:
handleMessages
are logged and then ignored, i.e. the processing moves forward in the partition.Dependencies:
Implementations:
Behavior:
handleMessages
throws an exception, then the messages are rejected and later reprocessed.Dependencies:
Implementations:
Behavior:
handleMessages
throws an exception, then the messages are rejected and later reprocessed.Dependencies:
Implementations:
Behavior:
handleMessages
throws an exception, then the messages are rejected and later reprocessed.Content
bytes are serialized to Base64 because SQS can only handle string content.Dependencies: