Chatter.MessageBrokers.AzureServiceBus 2.1.0

<a name="chatter-azureservicebus"></a> Chatter.MessageBrokers.AzureServiceBus

Azure Service Bus transport for the technology-agnostic Chatter.MessageBrokers abstractions.

Overview

Chatter.MessageBrokers.AzureServiceBus is the Azure Service Bus (ASB) implementation of the broker-agnostic interfaces defined in Chatter.MessageBrokers. It plugs an ASB sender and receiver into the messaging infrastructure so that messages dispatched and handled through your Chatter.CQRS command/event handlers flow over Azure Service Bus queues and topic subscriptions.

The core Chatter.MessageBrokers package registers the broker abstraction via IChatterBuilder.AddMessageBrokers(...). This package adds an AddAzureServiceBus(...) extension on IChatterBuilder that wires the concrete ASB sending/receiving components and ServiceBusOptions, registering an IMessagingInfrastructure keyed to the ASBMessageContext.InfrastructureType.

Key components registered:

• ServiceBusReceiver / ServiceBusReceiverFactory — pulls messages from queues and topic subscriptions.
• ServiceBusMessageSender / ServiceBusMessageSenderFactory / BrokeredMessageSenderPool — sends/publishes outbound messages.
• AzureServiceBusEntityPathBuilder — resolves queue/topic/subscription/rule paths (IBrokeredMessagePathBuilder).
• ServiceBusRetryExceptionPredicatesProvider / ServiceBusCircuitBreakerExceptionPredicatesProvider — feed ASB transient-exception detection into the Chatter retry and circuit-breaker recovery policies.

Installation

dotnet add package Chatter.MessageBrokers.AzureServiceBus

Getting Started

Register Chatter CQRS, the message broker abstraction, and then the Azure Service Bus transport. AddAzureServiceBus is chained off the IChatterBuilder returned by AddMessageBrokers:

using Microsoft.Extensions.DependencyInjection;

services
    .AddChatterCqrs(configuration)
    .AddMessageBrokers()
    .AddAzureServiceBus(asb =>
    {
        // connection can come from configuration (see Configuration below) or be set explicitly
        asb.WithConnectionString(configuration.GetConnectionString("ServiceBus"));
        asb.WithMaxConcurrentCalls(5);
        asb.WithPrefetchCount(10);

        // register the queues/subscriptions to receive from
        asb.AddQueueReceiver<CreateOrder>("orders-queue");
        asb.AddTopicSubscription<OrderCreated>("order-events-topic", "order-created-subscription");
    });

If WithConnectionString is omitted, the builder reads ServiceBusOptions from configuration (default section Chatter:Infrastructure:AzureServiceBus) — see Configuration. A connection string from either source is required; otherwise Build() throws.

Receiving

Receivers are registered while configuring options:

// commands -> queue (TMessage : ICommand)
asb.AddQueueReceiver<CreateOrder>(
    queueName: "orders-queue",
    errorQueuePath: "orders-error",
    transactionMode: TransactionMode.ReceiveOnly,
    maxReceiveAttempts: 10);

// events -> topic subscription (TMessage : IEvent)
asb.AddTopicSubscription<OrderCreated>(
    topicName: "order-events-topic",
    subscriptionName: "order-created-subscription",
    maxReceiveAttempts: 10);

Received messages are dispatched to the matching Chatter.CQRS handler:

public class CreateOrderHandler : IMessageHandler<CreateOrder>
{
    public Task Handle(CreateOrder message, IMessageHandlerContext context)
    {
        // handle the command
        return Task.CompletedTask;
    }
}

Sending

Within a handler you can reach ASB-specific send/publish/forward operations via the AzureServiceBus() extension on IMessageHandlerContext, which returns an IAzureServiceBusContextDispatcher:

using Chatter.CQRS.Context;

public class OrderCreatedHandler : IMessageHandler<OrderCreated>
{
    public async Task Handle(OrderCreated message, IMessageHandlerContext context)
    {
        await context.AzureServiceBus()
                     .Publish(new OrderShipped { OrderId = message.OrderId });
    }
}

IAzureServiceBusContextDispatcher composes the broker abstractions IMessageBrokerContextPublisher, IMessageBrokerContextSender, and IMessageBrokerContextForwarder.

Configuration

ServiceBusOptions is bound from configuration when a connection string is not supplied in code. The default configuration section is Chatter:Infrastructure:AzureServiceBus (override with UseConfig("Your:Section")).

{
  "Chatter": {
    "Infrastructure": {
      "AzureServiceBus": {
        "ConnectionString": "Endpoint=sb://your-namespace.servicebus.windows.net/;SharedAccessKeyName=...;SharedAccessKey=...",
        "MaxConcurrentCalls": 1,
        "PrefetchCount": 0
      }
    }
  }
}

ServiceBusOptions properties:

ServiceBusOptionsBuilder methods

The AddAzureServiceBus(asb => ...) delegate exposes a ServiceBusOptionsBuilder:

Retry and Circuit Breaker (receiving)

Receive-side recovery is driven by the broker-agnostic retry and circuit-breaker policies in Chatter.MessageBrokers.Recovery. This package contributes ASB-aware transient-exception detection:

• ServiceBusRetryExceptionPredicatesProvider (IRetryExceptionPredicatesProvider)
• ServiceBusCircuitBreakerExceptionPredicatesProvider (ICircuitBreakerExceptionPredicatesProvider)

Both treat a ServiceBusException as transient when its IsTransient is true, or when its Reason is ServiceBusFailureReason.ServiceCommunicationProblem, ServiceBusFailureReason.ServiceBusy, or ServiceBusFailureReason.ServiceTimeout. The per-receiver maxReceiveAttempts (default 10) bounds redelivery attempts before a message is routed to its configured errorQueuePath.

Authentication

The connection string above uses SAS-based auth. Azure Active Directory (AAD) authentication — via an Azure.Core.TokenCredential (falling back to DefaultAzureCredential when no explicit credential is given) — is provided by the sibling package Chatter.MessageBrokers.AzureServiceBus.Auth. When a token credential is supplied (via AddTokenProvider(...)) and the connection string contains no SAS token/key, that credential is used to authenticate to the namespace.

Testing

Real-namespace cross-entity transaction tests

The FullAtomicityViaInfrastructure mode relies on Azure Service Bus cross-entity (multi-top-level-entity) transactions. The local Service Bus emulator cannot exercise these — it throws Local transactions cannot span multiple top-level entities — so the cross-entity atomic commit/rollback tests run only against a real Azure Service Bus namespace. They are tagged Category=RealNamespaceIntegration (deliberately not Category=Integration, so the emulator test lane never selects them).

These tests skip cleanly when no real namespace is configured, so a plain dotnet test stays green without any Azure resources.

Run locally: set the CHATTER_ASB_REAL_NAMESPACE_CONNECTION_STRING environment variable to a connection string for a real Azure Service Bus namespace. The string must carry the Manage claim, because the test fixture creates and deletes uniquely-named queues (per run) via the Service Bus administration client.

export CHATTER_ASB_REAL_NAMESPACE_CONNECTION_STRING="Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<key>"
dotnet test src/Chatter.MessageBrokers.AzureServiceBus/tests/Chatter.MessageBrokers.AzureServiceBus.Tests.csproj --filter 'Category=RealNamespaceIntegration'

When the variable is unset (or blank) the tests are skipped at discovery time.

Run in CI: the real-namespace-integration job in .github/workflows/ci.yml runs this lane. Configure a GitHub Actions repository secret named CHATTER_ASB_REAL_NAMESPACE_CONNECTION_STRING (Manage-claim connection string) for it to execute. Without the secret (forks, or repos that have not configured it) the job is a clean no-op and never fails CI.

Sessions

Azure Service Bus message sessions deliver messages sharing the same SessionId to a single receiver in strict FIFO order. Chatter surfaces this through the existing Group Id term: inbound SessionId appears as MessageContext.GroupId; outbound session addressing reuses SendOptions.WithGroupId. No new session-specific API is introduced on the core.

Session-enabled entities (queues or subscriptions with RequiresSession = true) must be provisioned externally. The adapter does not auto-create or auto-enable them, consistent with the module's no-auto-provision stance.

Registering a session-enabled receiver

Use AddSessionQueueReceiver for Commands and AddSessionTopicSubscription for Events in place of their non-session counterparts:

services
    .AddChatterCqrs(configuration)
    .AddMessageBrokers()
    .AddAzureServiceBus(asb =>
    {
        asb.WithConnectionString(configuration.GetConnectionString("ServiceBus"));

        // session-enabled queue (Commands)
        asb.AddSessionQueueReceiver<ProcessOrder>("orders-session-queue");

        // session-enabled topic subscription (Events)
        asb.AddSessionTopicSubscription<OrderPlaced>("order-events-topic", "order-placed-session-sub");
    });

Each registered receiver processes one session at a time, holding it for FIFO delivery and rolling to the next when it is drained or goes idle. To increase throughput, run additional receiver instances; there is no max-concurrent-sessions knob.

Reading the session id in a handler

The inbound SessionId is surfaced as MessageContext.GroupId. Handlers read it through the broker-agnostic GroupId property — no Azure-specific import is required:

public class ProcessOrderHandler : IMessageHandler<ProcessOrder>
{
    public Task Handle(ProcessOrder message, IMessageHandlerContext context)
    {
        var sessionId = context.BrokeredMessage?.GetBrokeredMessageDetail()?.GroupId;
        // use sessionId to correlate work within this session
        return Task.CompletedTask;
    }
}

Sending a message to a session

Set WithGroupId on SendOptions to route the outbound message to the target session:

public class DispatchOrderHandler : IMessageHandler<DispatchOrder>
{
    public async Task Handle(DispatchOrder message, IMessageHandlerContext context)
    {
        var options = new SendOptions()
            .WithGroupId(message.OrderId);  // sets ServiceBusMessage.SessionId

        await context.AzureServiceBus()
                     .Send(new ProcessOrder { OrderId = message.OrderId }, "orders-session-queue", options);
    }
}

WithGroupId alone is enough: the mapping only assigns ServiceBusMessage.PartitionKey when a non-empty partition key was explicitly supplied, letting SessionId stand in for it otherwise. An explicit partition key is optional; if set, it must equal the Group Id, and it has to be a separate statement — WithMessageContext returns RoutingOptions, not SendOptions, so it cannot terminate the fluent chain above:

options.WithMessageContext(ASBMessageContext.PartitionKey, message.OrderId);

A mismatched partition key throws ArgumentOutOfRangeException from the Azure SDK's ServiceBusMessage.PartitionKey setter — client-side, before the message ever reaches the broker.

Inbound context inheritance

A handler that sends or publishes through IMessageHandlerContext (for example context.AzureServiceBus().Send(...)) inherits the entire inbound message context. A message received from a session-stamped entity therefore emits an outbound ServiceBusMessage.SessionId equal to the inbound one, even if the handler's own SendOptions never called WithGroupId.

This is by design and is not special to Group Id: CorrelationId, Subject, ReplyTo, ReplyToSessionId, To, and TimeToLive are inherited the same way. It's load-bearing only when the destination is session-enabled or partitioned — on a plain queue or topic an inherited Group Id is an inert wire property, but on a partitioned destination it still affects partition affinity even when that destination is not session-enabled.

To opt out, either resolve IBrokeredMessageDispatcher and call the overload that takes a TransactionContext instead of an IMessageHandlerContext — that overload does not merge inbound context — or supply your own Group Id on the outbound options, since caller-supplied options win the merge.

Durable per-session state

During handler execution a handler can read, write, and clear durable session state stored on the Azure Service Bus entity for the currently held session:

using Chatter.CQRS.Context;

public class ProcessOrderHandler : IMessageHandler<ProcessOrder>
{
    public async Task Handle(ProcessOrder message, IMessageHandlerContext context)
    {
        // read existing state (null when no state has been set)
        var stateBytes = await context.GetSessionStateAsync();

        // compute and persist new state
        var newState = BinaryData.FromString($"last-processed:{message.OrderId}");
        await context.SetSessionStateAsync(newState);

        // clear state when the session is complete
        // await context.ClearSessionStateAsync();
    }
}

GetSessionStateAsync, SetSessionStateAsync, and ClearSessionStateAsync are extension methods on IMessageHandlerContext provided by this package. Invoking them while handling a message that was not received through a session-enabled receiver throws InvalidOperationException.

Tuning session behavior

Two ServiceBusOptions knobs control how long a session is held. Both support fluent-or-config, with the fluent call winning in either direction:

asb.AddSessionQueueReceiver<ProcessOrder>("orders-session-queue");
asb.WithSessionIdleTimeout(TimeSpan.FromSeconds(30));
asb.WithMaxSessionLockRenewalDuration(TimeSpan.FromMinutes(10));

Or via configuration:

{
  "Chatter": {
    "Infrastructure": {
      "AzureServiceBus": {
        "SessionIdleTimeout": "00:00:30",
        "MaxSessionLockRenewalDuration": "00:10:00"
      }
    }
  }
}

These knobs apply only to session-enabled receivers. Non-session receivers are unaffected.

Trace Context and the Azure SDK's Diagnostic-Id

Chatter's opt-in tracing writes the W3C traceparent header onto outbound messages, where it rides the application properties in both directions. This has one interop consequence worth knowing about before you turn tracing on.

The Azure Service Bus SDK stamps its own legacy Diagnostic-Id correlation identifier only when the message does not already carry a correlation identifier. Because Chatter writes traceparent, the SDK's own Diagnostic-Id stamping is expected to be suppressed on Chatter-sent messages.

Confidence note. The presence of both mechanisms in the shipped Azure.Messaging.ServiceBus / Azure.Core assemblies is verified; the short-circuit control flow itself is taken from the Azure SDK's published source and is not verified in this repository. Treat the mitigation below as the safe course if your correlation depends on Diagnostic-Id, and confirm against your own traces.

Mitigation for applications that rely on Diagnostic-Id-based correlation: enable the SDK's ActivitySource support so it reads traceparent instead of stamping and reading Diagnostic-Id:

AZURE_EXPERIMENTAL_ENABLE_ACTIVITY_SOURCE=true

// equivalent AppContext switch
AppContext.SetSwitch("Azure.Experimental.EnableActivitySource", true);

Set it at process start, before the first Azure SDK type is touched: the SDK is documented to read the switch once, though that caching is likewise not verified here. Applications that do not correlate on Diagnostic-Id need no change — the SDK's own tracing stays off by default either way, and Chatter neither suppresses nor namespaces it.

One further observation on a mixed trace: Chatter's broker-boundary spans use the OpenTelemetry semantic conventions pinned at v1.30.0 (messaging.operation.type), while Azure.Messaging.ServiceBus still emits the older messaging.operation spelling. Both are valid under their respective pins; Chatter deliberately emits one spelling per concept rather than both. See ADR-0010.

Domain Language

See the domain glossary for definitions of Service Bus Receiver, Session Queue Receiver, Session Topic Subscription, Service Bus Sender, Service Bus Options, Service Bus Retry, Service Bus Circuit Breaker, Session, Session State, and Group Id ↔ SessionId realization.

← All Chatter modules