WorkR.Triggers.AzureServiceBus 0.3.3

WorkR.Triggers.AzureServiceBus

Azure Service Bus trigger for WorkR. Processes messages from a queue or topic subscription and drives them through a composable worker pipeline.

Installation

dotnet add package WorkR.Triggers.AzureServiceBus

AddServiceBusWorker

Typed messages — queue (recommended)

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

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

        // process the order...

        await context.Args.CompleteMessageAsync(context.Args.Message, cancellationToken);
    }
}

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

Typed messages — topic subscription

builder.Services.AddServiceBusWorker<OrderCreated, OrderCreatedWorker>(
    sp => sp.GetRequiredService<ServiceBusClient>(),
    topicName: "orders",
    subscriptionName: "fulfillment");

Raw messages

Receive the raw ProcessMessageEventArgs without deserialisation:

public class RawWorker : IWorker<ServiceBusTriggerContext>
{
    public async Task ExecuteAsync(
        ServiceBusTriggerContext context,
        CancellationToken cancellationToken)
    {
        var args = context.Value;  // ProcessMessageEventArgs
        var body = args.Message.Body.ToString();

        await args.CompleteMessageAsync(args.Message, cancellationToken);
    }
}

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

Worker Contract

Workers must either return a completed task or throw — the trigger does not impose any settlement outcome. Message settlement (complete, abandon, dead-letter, defer) is the worker's responsibility via context.Args. If no settlement is performed the SDK's default behaviour applies (controlled by ServiceBusProcessorOptions.AutoCompleteMessages, which defaults to true).

ServiceBusTriggerContext

The context passed to your worker for each message.

Message settlement (complete, abandon, dead-letter, defer) is performed directly via Args:

await context.Args.CompleteMessageAsync(context.Args.Message, cancellationToken);
await context.Args.AbandonMessageAsync(context.Args.Message, cancellationToken: cancellationToken);
await context.Args.DeadLetterMessageAsync(context.Args.Message, cancellationToken: cancellationToken);

Custom Deserialiser

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

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

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

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

Processor Options

WorkR creates a ServiceBusProcessor using the Azure SDK's default ServiceBusProcessorOptions. Pass a configure delegate to override any of these:

builder.Services.AddServiceBusWorker<OrderCreated, OrderCreatedWorker>(
    sp => sp.GetRequiredService<ServiceBusClient>(),
    "orders",
    configure: options =>
    {
        options.MaxConcurrentCalls = 4;
        options.AutoCompleteMessages = false;
    });

See ServiceBusProcessorOptions for the full list of available options and their defaults.

Full Pipeline Control

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

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

Topic/subscription variant:

builder.Services.AddServiceBusWorker<OrderCreated>(
    sp => sp.GetRequiredService<ServiceBusClient>(),
    topicName: "orders",
    subscriptionName: "fulfillment",
    pipeline => pipeline
        .AddWorker<ValidateWorker, ValidatedOrder>()
        .AddWorker<PersistWorker>());

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