Repository navigation
ConsumerMethod
You can consume method expressions in two ways: with dedicated worker threads, or with a pool of threads that multiple queues share (ConsumerMethodAsync).
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
var queueConnection = new QueueConnection(queueName, connectionString);
using (var schedulerContainer = new SchedulerContainer())
{
using (var scheduler = schedulerContainer.CreateTaskScheduler())
{
var factory = schedulerContainer.CreateTaskFactory(scheduler);
scheduler.Start();
using (var queueContainer = new QueueContainer<QueueInit>())
{
using (var queue = queueContainer.CreateConsumerMethodQueueScheduler(queueConnection, factory))
{
queue.Configuration.Worker.WorkerCount = 4;
var notifications = new ConsumerQueueNotifications(
(n) => logger.LogError($"Processing has failed {Environment.NewLine}{n.Error}"),
(n) => logger.LogError($"Processing has failed to dequeue a message {Environment.NewLine}{n.Error}"),
(n) => logger.LogError($"Processing has failed {Environment.NewLine}{n.MessageId}{Environment.NewLine}{n.Error}"),
(n) => logger.LogError($"Processing has triggered a poison message {Environment.NewLine}{n.MessageId}{Environment.NewLine}{n.Error}"),
(n) => logger.LogWarning($"Processing has triggered a rollback {Environment.NewLine}{n.MessageId}{Environment.NewLine}{n.Error}"),
(n) => logger.LogInformation($"Processing completed {n.MessageId}"));
queue.Start(notifications);
Console.WriteLine("Processing messages - press any key to stop");
Console.ReadKey((true));
}
}
}
}Unlike the typed-message consumer, this consumer does not use a delegate for message processing. Instead, it executes the lambda expression. All types must be resolvable, or errors will be thrown.
You may want to check for the following condition in your code:
- 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();
}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 a complete working example, see SQLServerConsumerLinq in the samples repository.
For any issues please use the GitHub issues