redb.Route.AzureServiceBus 2.0.2

redb.Route.AzureServiceBus

Azure Service Bus connector for the redb.Route ESB framework.
Provides a full-featured producer (single & batch), consumer (PeekLock / ReceiveAndDelete), session consumer (FIFO per session), and transacted acknowledgement. Uses the asb URI scheme.

Quick Start

// Consumer: receive from a queue
.From(Asb.Queue("orders")
    .ConnectionString("Endpoint=sb://my-ns.servicebus.windows.net/;SharedAccessKeyName=...;SharedAccessKey=...")
    .MaxConcurrentCalls(5)
    .PrefetchCount(10))

// Producer: send to a queue
.To(Asb.Queue("orders")
    .ConnectionString("Endpoint=sb://..."))

// Topic/subscription roundtrip
.From(Asb.Topic("events", "my-subscription")
    .ConnectionString("Endpoint=sb://..."))

.To(Asb.Topic("events", "ignored-for-producer")
    .ConnectionString("Endpoint=sb://..."))

URI Format

asb://entity-name?connectionString=...&param=value

Entity detection is automatic: if subscriptionName is set, the entity is treated as a topic; otherwise as a queue.

Producer

The producer sends messages via ServiceBusSender. Two modes are supported:

Single Message (default)

var exchange = new Exchange(new Message("Hello ASB"));
exchange.In.Headers[AzureServiceBusHeaders.CorrelationId] = "corr-123";
exchange.In.Headers[AzureServiceBusHeaders.Subject] = "order.created";
await producer.Process(exchange);

// After send, MessageId header is set
var id = exchange.In.Headers[AzureServiceBusHeaders.MessageId];

Batch Mode

Enable with enableBatch=true. Body must be IEnumerable. Each item is serialized to BinaryData and added to a batch respecting batchMaxMessages and batchMaxSizeBytes.

.To(Asb.Queue("orders")
    .ConnectionString("...")
    .EnableBatch(true)
    .BatchMaxMessages(50)
    .BatchMaxSizeBytes(262144))

var items = new[] { "msg-1", "msg-2", "msg-3" };
var exchange = new Exchange(new Message(items));
await producer.Process(exchange);

var count = (int)exchange.In.Headers[AzureServiceBusHeaders.BatchMessageCount]!; // 3

Body Resolution

Producer Headers

Headers with the redbAsb. prefix are mapped to native ServiceBusMessage properties.
All other headers are copied to ApplicationProperties.

// Native ASB properties
exchange.In.Headers[AzureServiceBusHeaders.MessageId] = "custom-id";
exchange.In.Headers[AzureServiceBusHeaders.SessionId] = "session-1";
exchange.In.Headers[AzureServiceBusHeaders.ScheduledEnqueueTime] = DateTimeOffset.UtcNow.AddMinutes(5);
exchange.In.Headers[AzureServiceBusHeaders.TimeToLive] = TimeSpan.FromMinutes(30);

// Application properties (forwarded as-is)
exchange.In.Headers["X-Trace-Id"] = "abc-123";
exchange.In.Headers["X-Priority"] = "high";

Note: PartitionKey is only set when SessionId is empty (ASB constraint).

Consumer

The consumer uses ServiceBusProcessor with callback-based message delivery.

.From(Asb.Queue("orders")
    .ConnectionString("...")
    .ReceiveMode("PeekLock")          // default
    .MaxConcurrentCalls(10)
    .PrefetchCount(20)
    .MaxAutoLockRenewalDuration(300))  // 5 min

Acknowledge Modes

// Auto dead-letter on processing error
.From(Asb.Queue("orders")
    .ConnectionString("...")
    .AutoDeadLetter(true)
    .DeadLetterReason("ProcessingFailed"))

ReceiveAndDelete

Messages are removed from the queue immediately upon receipt. No acknowledgement needed:

.From(Asb.Queue("orders")
    .ConnectionString("...")
    .ReceiveMode("ReceiveAndDelete"))

Sub-Queues

Read from dead-letter or transfer dead-letter queues:

.From(Asb.Queue("orders")
    .ConnectionString("...")
    .SubQueue("deadletter"))

Consumer Headers

Each received message populates the following exchange headers:

All ApplicationProperties from the message are also copied to exchange headers.

Session Consumer

Enable session-aware FIFO processing with enableSessions=true:

.From(Asb.Queue("session-queue")
    .ConnectionString("...")
    .EnableSessions(true)
    .MaxConcurrentSessions(5)
    .SessionIdleTimeout(30))

// Filter by specific session
.From(Asb.Queue("session-queue")
    .ConnectionString("...")
    .EnableSessions(true)
    .SessionId("my-session-id"))

The session consumer uses ServiceBusSessionProcessor and guarantees message ordering within a session. MaxConcurrentSessions controls parallelism across sessions; messages within the same session are always processed sequentially.

Transacted Mode

When transacted=true, message acknowledgement is deferred. The consumer registers ITransactedAction instances in exchange.Properties["TRANSACT_ACTION"] — a ConcurrentDictionary<string, ITransactedAction>:

.From(Asb.Queue("orders")
    .ConnectionString("...")
    .Transacted(true))

// In your processor:
var actions = exchange.Properties["TRANSACT_ACTION"]
    as ConcurrentDictionary<string, ITransactedAction>;

// On success:
foreach (var action in actions!.Values)
    await action.Commit();   // CompleteMessageAsync

// On failure:
foreach (var action in actions!.Values)
    await action.Rollback(); // AbandonMessageAsync

Topic / Subscription

// Producer sends to the topic
.To(Asb.Topic("events", "sub-name")
    .ConnectionString("..."))

// Consumer reads from a subscription
.From(Asb.Topic("events", "my-subscription")
    .ConnectionString("..."))

The second argument to Asb.Topic() is the subscription name. The producer ignores subscriptionName and sends directly to the topic; the consumer uses it to create a ServiceBusProcessor for that subscription.

Connection Factory

For complex or shared client configurations, register a named AzureServiceBusConnectionFactory:

var factory = new AzureServiceBusConnectionFactory
{
    ConnectionString = "Endpoint=sb://my-ns.servicebus.windows.net/;...",
    MaxRetries = 5,
    RetryMode = ServiceBusRetryMode.Fixed,
    TransportType = ServiceBusTransportType.AmqpWebSockets,
    ProxyAddress = "http://proxy:8080",
};

context.AddToRegistry("myAsb", factory);

// Reference by name in URI
Asb.Queue("orders").ConnectionFactory("myAsb")

If ProxyAddress is set, transport is automatically switched to AmqpWebSockets with the configured proxy.

Factory Properties

DI Registration

services.AddRedbRouteAzureServiceBus();

Configuration Reference

Connection

Consumer

Sessions

Producer

Batch

Retry

Headers Reference

All headers use the redbAsb. prefix.

Requirements

• Azure Service Bus or Azure Service Bus Emulator
• .NET 8.0 / 9.0 / 10.0
• Azure.Messaging.ServiceBus 7.x