WorkR.Triggers.AzureStorageQueues 0.3.3

WorkR.Triggers.AzureStorageQueues

Azure Storage Queue polling trigger for WorkR. Polls a queue, [optionally] deserialises messages, and passes them through a composable worker pipeline.

Installation

dotnet add package WorkR.Triggers.AzureStorageQueues

AddStorageQueueWorker

Typed messages (recommended)

Deserialise queue messages to a strongly-typed model before passing them to your worker. JSON deserialisation is used by default.

public class OrderCreatedWorker : IWorker<StorageQueueTriggerContext<OrderCreated>>
{
    public async Task ExecuteAsync(
        StorageQueueTriggerContext<OrderCreated> context,
        CancellationToken cancellationToken)
    {
        var order = context.Value;  // deserialized OrderCreated

        await ProcessAsync(order, cancellationToken);

        await context.DeleteMessageAsync(cancellationToken);
    }
}

builder.Services.AddStorageQueueWorker<OrderCreated, OrderCreatedWorker>(
    sp => sp.GetRequiredService<QueueServiceClient>(),
    "orders");

Raw messages

Receive the raw QueueMessage without deserialisation:

public class RawWorker : IWorker<StorageQueueTriggerContext>
{
    public async Task ExecuteAsync(
        StorageQueueTriggerContext context,
        CancellationToken cancellationToken)
    {
        var body = context.Value.Body.ToString();  // raw message body

        await ProcessAsync(body, cancellationToken);

        await context.DeleteMessageAsync(cancellationToken);
    }
}

builder.Services.AddStorageQueueWorker<RawWorker>(
    sp => sp.GetRequiredService<QueueServiceClient>(),
    "my-queue");

StorageQueueTriggerContext

The context passed to your worker for each message.

Worker Contract

Each message is processed on the thread pool independently of the polling loop. Workers must either return a completed task or throw — the trigger handles settlement based on the outcome and the configured options.

Message Completion

By default (AutoCompleteMessages = true) the trigger deletes the message automatically after the worker returns successfully. If the worker throws, the message is left on the queue and will become visible again after VisibilityTimeout. Once the message has been dequeued MaxDeliveryCount times it is automatically moved to the poison queue.

To take manual control, set AutoCompleteMessages = false and call DeleteMessageAsync or DeadLetterMessageAsync yourself:

builder.Services.AddStorageQueueWorker<OrderCreated, OrderCreatedWorker>(
    sp => sp.GetRequiredService<QueueServiceClient>(),
    "orders",
    configure: options => options.AutoCompleteMessages = false);

public async Task ExecuteAsync(StorageQueueTriggerContext<OrderCreated> context, CancellationToken cancellationToken)
{
    try
    {
        await ProcessAsync(context.Value, cancellationToken);
        await context.DeleteMessageAsync(cancellationToken);
    }
    catch (NonRetryableException)
    {
        await context.DeadLetterMessageAsync(cancellationToken);
    }
}

Custom Deserialiser

Supply a custom deserialiser to control how message bodies are converted to your model:

builder.Services.AddStorageQueueWorker<OrderCreated, OrderCreatedWorker>(
    sp => sp.GetRequiredService<QueueServiceClient>(),
    "orders",
    deserializerFactory: sp => async message =>
    {
        var json = message.Body.ToString();
        return JsonSerializer.Deserialize<OrderCreated>(json, myOptions)!;
    });

The built-in JSON deserialiser can also be configured with custom JsonSerializerOptions:

deserializerFactory: _ => StorageQueueMessageDeserializers.Json<OrderCreated>(myJsonOptions)

StorageQueueTriggerOptions

MaxConcurrentCalls and MaxMessages are independent. Set MaxMessages to control how many messages are fetched in one network call; set MaxConcurrentCalls to control how many workers run in parallel.

PollingDelay and ErrorDelay accept a StorageQueueDelay delegate — use StorageQueueDelayStrategy for built-in strategies:

configure: options =>
{
    options.PollingDelay = StorageQueueDelayStrategy.Fixed(TimeSpan.FromSeconds(10));
    options.MaxMessages = 8;
}

Full Pipeline Control

Use the builder overload to chain multiple workers or configure per-step middleware:

builder.Services.AddStorageQueueWorker<OrderCreated>(
    sp => sp.GetRequiredService<QueueServiceClient>(),
    "orders",
    pipeline => pipeline
        .AddWorker<ValidateWorker, ValidatedOrder>()
        .AddWorker<PersistWorker>());

Default middleware: UseScope. Applied to the first worker in the chain only.