Repository navigation
ConsumerMethodAsync
You can consume messages in two ways: with dedicated worker threads (ConsumerMethod), or with a pool of threads that multiple queues share.
You need the following:
- A task scheduler instance
- 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
var queueConnection = new QueueConnection(queueName, connectionString);
using (var schedulerContainer = new SchedulerContainer())
{
using (var scheduler = schedulerContainer.CreateTaskScheduler())
{
var factory = schedulerContainer.CreateTaskFactory(scheduler);
factory.Scheduler.Configuration.MaximumThreads = 8;
factory.Scheduler.Configuration.MinimumThreads = 0;
factory.Scheduler.Start();
using (var queueContainer = new QueueContainer<QueueInit>())
{
using (
var queue = queueContainer.CreateConsumerMethodQueueScheduler(queueConnection, factory)
)
{
var notifications = new ConsumerQueueNotifications(
(n) => Console.WriteLine($"Error: {n.Error}"),
(n) => Console.WriteLine($"Receive error: {n.Error}"),
(n) => Console.WriteLine($"Moved to error queue: {n.MessageId}"),
(n) => Console.WriteLine($"Poison message: {n.MessageId}"),
(n) => Console.WriteLine($"Rollback: {n.MessageId}"),
(n) => Console.WriteLine($"Completed: {n.MessageId}"));
queue.Start(notifications);
Console.WriteLine("Processing messages - press any key to stop");
Console.ReadKey((true));
}
}
}
}If you do not pass a scheduler and factory to the creation logic, a default scheduler and factory are used. But then you cannot change their configuration. You also cannot re-use the scheduler with multiple queues.
When your method expression finishes with no errors, the expression 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.
To check these flags, you must pass the IWorkerNotifications instance to your method expression.
i.e.
var result = queue.Send((m, w) => Console.WriteLine(w.TransportSupportsRollback));if (w.TransportSupportsRollback && w.MessageCancellation.Token.IsCancellationRequested)
{
logger.LogDebug("Cancel has been requested - aborting");
w.MessageCancellation.Token.ThrowIfCancellationRequested();
}Instead, you can register a delegate with the cancel token. Then your code does not have to check the token over and over.
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. If you created a task scheduler, you should dispose of it after you dispose of all queues.
scheduler.Dispose();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 a complete working example, see SQLServerConsumerLinq in the samples repository.
For any issues please use the GitHub issues