Repository navigation
Consumer
You can consume messages in two ways: with dedicated worker threads, or with a pool of threads that multiple queues share (ConsumerAsync).
You need the following:
- The transport that you connect to. The transport init module that you specify when you create the container determines the transport.
- The connection string to the transport
- The name of the queue
- A delegate that the queue calls for each message to process
ConsumerQueueNotifications provides 6 callback delegates for error handling and completion tracking.
using (var queueContainer = new QueueContainer<QueueInit>())
{
var queueConnection = new QueueConnection(queueName, connectionString);
using (var queue = queueContainer.CreateConsumer(queueConnection))
{
queue.Configuration.Worker.WorkerCount = 4;
var notifications = new ConsumerQueueNotifications(
(notification) => Console.WriteLine($"Error: {notification.Error}"),
(notification) => Console.WriteLine($"Receive error: {notification.Error}"),
(notification) => Console.WriteLine($"Moved to error queue: {notification.MessageId}"),
(notification) => Console.WriteLine($"Poison message: {notification.MessageId}"),
(notification) => Console.WriteLine($"Rollback: {notification.MessageId}"),
(notification) => Console.WriteLine($"Completed: {notification.MessageId}"));
queue.Start<SimpleMessage>(HandleMessages, notifications);
Console.WriteLine("Processing messages - press any key to stop");
Console.ReadKey((true));
}
}
private void HandleMessages(IReceivedMessage<SimpleMessage> message, IWorkerNotification notifications)
{
//processing logic goes here
}When your delegate method finishes with no errors, the message is acked and considered finished. But you may want to check for a condition in your logic:
- The queue shutting down. A cancel token is provided for this.
- If the transport supports rollback, you can throw an operation canceled exception to requeue the message
For example, the code below checks if cancellation is requested and forces a requeue. Before it does this, it confirms that the transport supports rollbacks.
if (notifications.TransportSupportsRollback)
{
var canceled = notifications.MessageCancellation.Token.WaitHandle.WaitOne(
TimeSpan.FromMilliseconds(processingTime));
if (canceled) throw new OperationCanceledException("Processing was canceled");
}Instead, you can register a delegate with the cancel token. Then your code does not have to check the token over and over.
Your handler receives an IWorkerNotification with these properties:
| Property | Type | Description |
|---|---|---|
WorkerStopping |
ICancelWork |
Fires when the queue is shutting down. Check this to stop long-running work cooperatively. |
HeaderNames |
IHeaders |
Well-known header names |
HeartBeat |
IWorkerHeartBeatNotification |
Heartbeat status for the current message |
TransportSupportsRollback |
bool |
Whether the transport supports rolling back a de-queue. Always check this before you throw OperationCanceledException. |
Log |
ILogger |
Microsoft.Extensions.Logging logger scoped to the worker |
Metrics |
IMetrics |
Metrics logging |
Tracer |
ActivitySource |
OpenTelemetry ActivitySource for distributed tracing |
MessageCancellation |
IMessageCancellation |
Per-message cancellation token. Fires on graceful shutdown or a dashboard-initiated cancel |
Use the standard MEL extension methods on notifications.Log:
notifications.Log.LogDebug($"Starting message {message.MessageId.Id.Value}");
notifications.Log.LogInformation($"Processing complete");
notifications.Log.LogWarning($"Retrying due to transient error");
notifications.Log.LogError($"Unhandled exception during processing");For a complete working example, see SQLServerConsumer in the samples repository.
To stop the queue, call dispose on it.
queue.Dispose();If you dispose of the queue container, it also disposes all queues that it created.
Dispose is a blocking operation. It may take a while to return. The time depends on your configuration settings and on how quickly your message consuming code responds to cancels.
For any issues please use the GitHub issues