ConsumerAsync

You can consume messages in two ways: with dedicated worker threads (Consumer), or with a pool of threads that multiple queues share.

This page shows the scheduler-based async consumer (CreateConsumerQueueScheduler). It manages a thread pool through TaskScheduler. The lower-level CreateConsumerAsync API uses the same Start() pattern, but the worker action must return Task.

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
• A delegate that the queue calls for each message to process

ConsumerQueueNotifications provides 6 callback delegates for error handling and completion tracking.

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<SqlServerMessageQueueInit>())
        {
            var queueConnection = new QueueConnection(queueName, connectionString);
            using (var queue = queueContainer.CreateConsumerQueueScheduler(queueConnection, factory))
            {
                var notifications = new ConsumerQueueNotifications(
                    (notification) => { /* handle errors */ },
                    (notification) => { /* handle receive errors */ },
                    (notification) => { /* handle message moved to error queue */ },
                    (notification) => { /* handle poison message */ },
                    (notification) => { /* handle rollback */ },
                    (notification) => { /* handle completion */ });
                queue.Start<SimpleMessage>(Handle, notifications);
                Console.WriteLine("Processing messages - press any key to stop");
                Console.ReadKey((true));
            }
        }
    }
}

private void Handle(IReceivedMessage<SimpleMessage> m, IWorkerNotification n)
{
    //processing logic goes here
}

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 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 && notifications.MessageCancellation.Token.IsCancellationRequested)
{
    notifications.Log.LogDebug("Cancel has been requested - aborting");
    notifications.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 SQLServerConsumerAsync in the samples repository.