From fc1468626b11c5d6ebca312bc845954c876ce325 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 20 Jul 2026 11:34:10 -0500 Subject: [PATCH 1/2] GH-3533: pin Azure Service Bus session listeners to specific session identifiers (5.x backport) Backport of the 6.x change to the 5.0 maintenance line, bumped to 5.40.0. Add an opt-in ServiceBusSessionProcessor-based session listener so competing consumers on ONE shared queue/subscription can each be pinned to their own session id(s). On a shared entity the session id becomes a broker-enforced routing key, so a listener pinned to "A" never sees the messages meant for "B". - ConfigureSessionProcessor(Action) on both queue and subscription listener configs (multicast, so it composes) - RequireSessionsWithOnlyTheseIdentifiers(params string[]) sugar that populates ServiceBusSessionProcessorOptions.SessionIds - BuildSessionProcessorOptions maps ListenerCount -> MaxConcurrentSessions, keeps in-session FIFO, and reserves ReceiveMode=PeekLock / AutoCompleteMessages=false - New InlineAzureServiceBusSessionListener; AzureServiceBusEnvelope gains a ProcessSessionMessageEventArgs complete/defer/dead-letter path Gated: the legacy AcceptNextSession loop stays the default and only flips to the processor when ConfigureSessionProcessor is set, so existing session listeners are unaffected. Adapted to the 5.0 code shape (no ConfigureProcessor / IReportConnectionState / PrefetchCount that exist on 6.x). No conditional compilation needed: the Azure SDK session-processor APIs are available on net8.0/net9.0/net10.0. Unit tests + an emulator end-to-end test reproducing the issue, all green on net9.0; full project builds clean on all three TFMs. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01FKAxzuZ36VP6UPcTQf3MUs --- Directory.Build.props | 2 +- .../configuring_session_processor_options.cs | 133 ++++++++++++ .../session_id_pinning.cs | 90 +++++++++ ...ureServiceBusQueueListenerConfiguration.cs | 46 +++++ ...iceBusSubscriptionListenerConfiguration.cs | 46 +++++ .../AzureServiceBusTransport.Listening.cs | 59 ++++++ .../Internal/AzureServiceBusEndpoint.cs | 13 ++ .../Internal/AzureServiceBusEnvelope.cs | 16 +- .../InlineAzureServiceBusSessionListener.cs | 189 ++++++++++++++++++ 9 files changed, 592 insertions(+), 2 deletions(-) create mode 100644 src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs create mode 100644 src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_id_pinning.cs create mode 100644 src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusSessionListener.cs diff --git a/Directory.Build.props b/Directory.Build.props index 3a993ae72..f51b23584 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -11,7 +11,7 @@ 1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618;VSTHRD200 true enable - 5.39.5 + 5.40.0 $(PackageProjectUrl) true true diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs new file mode 100644 index 000000000..e91f34855 --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs @@ -0,0 +1,133 @@ +using Azure.Messaging.ServiceBus; +using JasperFx.Core; +using Shouldly; +using Wolverine.AzureServiceBus.Internal; +using Wolverine.Configuration; +using Xunit; + +namespace Wolverine.AzureServiceBus.Tests; + +public class configuring_session_processor_options +{ + [Fact] + public void configure_session_processor_stores_the_action_and_requires_sessions_on_the_queue() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + configuration.ConfigureSessionProcessor(o => o.MaxConcurrentSessions = 4); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + queue.ConfigureSessionProcessor.ShouldNotBeNull(); + queue.Options.RequiresSession.ShouldBeTrue(); + } + + [Fact] + public void configure_session_processor_stores_the_action_and_requires_sessions_on_the_subscription() + { + var transport = new AzureServiceBusTransport(); + var topic = transport.Topics["topic1"]; + var subscription = topic.FindOrCreateSubscription("sub1"); + + var configuration = new AzureServiceBusSubscriptionListenerConfiguration(subscription); + configuration.ConfigureSessionProcessor(o => o.MaxConcurrentSessions = 4); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + subscription.ConfigureSessionProcessor.ShouldNotBeNull(); + subscription.Options.RequiresSession.ShouldBeTrue(); + } + + [Fact] + public void require_sessions_with_only_these_identifiers_populates_session_ids() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + configuration.RequireSessionsWithOnlyTheseIdentifiers("A", "B"); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + queue.Options.RequiresSession.ShouldBeTrue(); + + var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue); + options.SessionIds.ShouldBe(new[] { "A", "B" }); + } + + [Fact] + public void build_session_processor_options_reasserts_peek_lock_and_disables_autocomplete() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + // A user trying to break Wolverine's acknowledgement contract must not win + queue.ConfigureSessionProcessor = o => + { + o.ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete; + o.AutoCompleteMessages = true; + }; + + var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue); + + options.ReceiveMode.ShouldBe(ServiceBusReceiveMode.PeekLock); + options.AutoCompleteMessages.ShouldBeFalse(); + } + + [Fact] + public void build_session_processor_options_maps_listener_count_to_max_concurrent_sessions() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + configuration.RequireSessions(8).ConfigureSessionProcessor(_ => { }); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue); + options.MaxConcurrentSessions.ShouldBe(8); + + // FIFO ordering per session is preserved + options.MaxConcurrentCallsPerSession.ShouldBe(1); + } + + [Fact] + public void build_session_processor_options_composes_multiple_actions() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + + // The SessionIds sugar and an explicit customization must both apply + configuration + .RequireSessionsWithOnlyTheseIdentifiers("only-me") + .ConfigureSessionProcessor(o => o.MaxAutoLockRenewalDuration = 10.Minutes()); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue); + + options.SessionIds.ShouldBe(new[] { "only-me" }); + options.MaxAutoLockRenewalDuration.ShouldBe(10.Minutes()); + } + + [Fact] + public void session_listener_defaults_to_the_legacy_loop_when_no_customization() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + configuration.RequireSessions(); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + // Zero behavior change for existing session users: the processor path is opt-in only + queue.Options.RequiresSession.ShouldBeTrue(); + queue.ConfigureSessionProcessor.ShouldBeNull(); + } +} diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_id_pinning.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_id_pinning.cs new file mode 100644 index 000000000..c6cc544c2 --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_id_pinning.cs @@ -0,0 +1,90 @@ +using Azure.Messaging.ServiceBus; +using IntegrationTests; +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.Tracking; +using Xunit; + +namespace Wolverine.AzureServiceBus.Tests; + +// GH-3533: pinning a session-enabled listener to specific session identifiers turns the session id +// into a broker-enforced routing key on a shared queue, so a listener pinned to "A" never sees the +// messages meant for "B". +[Trait("Category", "Flaky")] +public class session_id_pinning : IAsyncLifetime +{ + private IHost _host = null!; + + public async Task InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.UseAzureServiceBusTesting().AutoProvision().AutoPurgeOnStartup(); + + opts.ListenToAzureServiceBusQueue("shared-pinned") + + // Only ever lock the "A" session on this shared queue + .RequireSessionsWithOnlyTheseIdentifiers("A") + .Sequential(); + + opts.PublishMessage().ToAzureServiceBusQueue("shared-pinned"); + }).StartAsync(); + } + + public async Task DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + await AzureServiceBusTesting.DeleteAllEmulatorObjectsAsync(); + } + + [Fact] + public async Task pinned_listener_only_receives_its_own_session() + { + await using var client = new ServiceBusClient(Servers.AzureServiceBusConnectionString); + + // Seed a message destined for session "B" directly onto the shared queue, bypassing Wolverine + var sender = client.CreateSender("shared-pinned"); + await sender.SendMessageAsync(new ServiceBusMessage("not for A") + { + SessionId = "B", + MessageId = Guid.NewGuid().ToString() + }); + + // Drive three "A" messages through Wolverine and confirm ONLY those are received + Func sendAll = async bus => + { + await bus.SendAsync(new PinnedMessage("A-1"), new DeliveryOptions { GroupId = "A" }); + await bus.SendAsync(new PinnedMessage("A-2"), new DeliveryOptions { GroupId = "A" }); + await bus.SendAsync(new PinnedMessage("A-3"), new DeliveryOptions { GroupId = "A" }); + }; + + var tracked = await _host.TrackActivity() + .IncludeExternalTransports() + .Timeout(30.Seconds()) + .ExecuteAndWaitAsync(sendAll); + + // Every "A" message was delivered here (order is a separate FIFO concern), and nothing else + tracked.Received.MessagesOf().Select(x => x.Name).OrderBy(x => x) + .ShouldBe(["A-1", "A-2", "A-3"]); + + // The "B" session message must still be sitting on the shared queue, never delivered to the + // A-pinned listener. + await using var sessionReceiver = await client.AcceptSessionAsync("shared-pinned", "B"); + var leftover = await sessionReceiver.ReceiveMessageAsync(5.Seconds()); + leftover.ShouldNotBeNull(); + leftover.SessionId.ShouldBe("B"); + } +} + +public record PinnedMessage(string Name); + +public static class PinnedMessageHandler +{ + public static void Handle(PinnedMessage message) + { + // no-op; tracking observes receipt + } +} diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs index 580cc5394..2fe230ee1 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs @@ -1,3 +1,4 @@ +using Azure.Messaging.ServiceBus; using Azure.Messaging.ServiceBus.Administration; using Wolverine.AzureServiceBus.Internal; using Wolverine.Configuration; @@ -142,6 +143,51 @@ public AzureServiceBusQueueListenerConfiguration RequireSessions(int? listenerCo return this; } + /// + /// Customize the Azure Service Bus used by this + /// session-enabled listener -- e.g. MaxConcurrentSessions, MaxAutoLockRenewalDuration, + /// SessionIdleTimeout, or SessionIds. Calling this implies + /// and switches the session listener from the default AcceptNextSession loop to a + /// . Multiple calls compose. Wolverine reserves control of the + /// properties it depends on for message acknowledgement (ReceiveMode, AutoCompleteMessages), + /// which are re-asserted after this action runs. + /// + /// + /// + public AzureServiceBusQueueListenerConfiguration ConfigureSessionProcessor( + Action configure) + { + add(e => + { + e.Options.RequiresSession = true; + // Compose rather than overwrite so the SessionIds sugar can coexist with an explicit hook + e.ConfigureSessionProcessor += configure; + }); + return this; + } + + /// + /// Pin this listener to only the given session identifiers. On a shared queue this turns the session + /// id into a broker-enforced routing key: competing consumers each pinned to their own id(s) never see + /// each other's messages. Producers select the target by setting DeliveryOptions.GroupId to the + /// session id. Delegates to by populating + /// ServiceBusSessionProcessorOptions.SessionIds. (GH-3533) + /// + /// The session identifiers this listener should exclusively lock + /// + public AzureServiceBusQueueListenerConfiguration RequireSessionsWithOnlyTheseIdentifiers( + params string[] identifiers) + { + RequireSessions(); + return ConfigureSessionProcessor(options => + { + foreach (var id in identifiers) + { + options.SessionIds.Add(id); + } + }); + } + /// /// Utilize custom envelope mapping for Amazon Service Bus interoperability with external non-Wolverine systems /// diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs index 94e1eaedc..a23ddce66 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs @@ -1,3 +1,4 @@ +using Azure.Messaging.ServiceBus; using Azure.Messaging.ServiceBus.Administration; using Wolverine.AzureServiceBus.Internal; using Wolverine.Configuration; @@ -128,6 +129,51 @@ public AzureServiceBusSubscriptionListenerConfiguration RequireSessions(int? lis return this; } + /// + /// Customize the Azure Service Bus used by this + /// session-enabled subscription listener -- e.g. MaxConcurrentSessions, + /// MaxAutoLockRenewalDuration, SessionIdleTimeout, or SessionIds. Calling this + /// implies and switches the session listener from the default + /// AcceptNextSession loop to a . Multiple calls compose. + /// Wolverine reserves control of the properties it depends on for message acknowledgement + /// (ReceiveMode, AutoCompleteMessages), which are re-asserted after this action runs. + /// + /// + /// + public AzureServiceBusSubscriptionListenerConfiguration ConfigureSessionProcessor( + Action configure) + { + add(e => + { + e.Options.RequiresSession = true; + // Compose rather than overwrite so the SessionIds sugar can coexist with an explicit hook + e.ConfigureSessionProcessor += configure; + }); + return this; + } + + /// + /// Pin this listener to only the given session identifiers. On a shared subscription this turns the + /// session id into a broker-enforced routing key: competing consumers each pinned to their own id(s) + /// never see each other's messages. Producers select the target by setting DeliveryOptions.GroupId + /// to the session id. Delegates to by populating + /// ServiceBusSessionProcessorOptions.SessionIds. (GH-3533) + /// + /// The session identifiers this listener should exclusively lock + /// + public AzureServiceBusSubscriptionListenerConfiguration RequireSessionsWithOnlyTheseIdentifiers( + params string[] identifiers) + { + RequireSessions(); + return ConfigureSessionProcessor(options => + { + foreach (var id in identifiers) + { + options.SessionIds.Add(id); + } + }); + } + /// /// Utilize custom envelope mapping for Amazon Service Bus interoperability with external non-Wolverine systems /// diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs index 21c3b059e..9fd38f0cf 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs @@ -60,6 +60,23 @@ private async Task buildListenerForQueue(IWolverineRuntime runtime, I if (queue.Options.RequiresSession) { + // GH-3533: when the endpoint carries any ServiceBusSessionProcessorOptions customization + // (most importantly SessionIds pinning), use the SDK's ServiceBusSessionProcessor instead + // of the default AcceptNextSession loop. Gated so current session listeners are unchanged. + if (queue.ConfigureSessionProcessor != null) + { + var sessionProcessor = + BusClient.CreateSessionProcessor(queue.QueueName, BuildSessionProcessorOptions(queue)); + + var sessionListener = new InlineAzureServiceBusSessionListener(queue, + runtime.LoggerFactory.CreateLogger(), sessionProcessor, + receiver, mapper, requeue); + + await sessionListener.StartAsync(); + + return sessionListener; + } + return new AzureServiceBusSessionListener(this, queue, receiver, mapper, runtime.LoggerFactory.CreateLogger(), requeue); } @@ -116,6 +133,22 @@ private async Task buildListenerForSubscription(IWolverineRuntime run if (subscription.Options.RequiresSession) { + // GH-3533: see buildListenerForQueue -- the session processor path is opt-in via + // ConfigureSessionProcessor (e.g. RequireSessionsWithOnlyTheseIdentifiers). + if (subscription.ConfigureSessionProcessor != null) + { + var sessionProcessor = BusClient.CreateSessionProcessor(subscription.Topic.TopicName, + subscription.SubscriptionName, BuildSessionProcessorOptions(subscription)); + + var sessionListener = new InlineAzureServiceBusSessionListener(subscription, + runtime.LoggerFactory.CreateLogger(), sessionProcessor, + receiver, mapper, requeue); + + await sessionListener.StartAsync(); + + return sessionListener; + } + return new AzureServiceBusSessionListener(this, subscription, receiver, mapper, runtime.LoggerFactory.CreateLogger(), requeue); } @@ -138,4 +171,30 @@ private async Task buildListenerForSubscription(IWolverineRuntime run return listener; } + + // Builds the ServiceBusSessionProcessorOptions for the opt-in ServiceBusSessionProcessor session + // listener (GH-3533). Applies the user's (multicast) customization -- including any SessionIds + // pinning -- then re-asserts the acknowledgement properties Wolverine's + // InlineAzureServiceBusSessionListener depends on. + internal static ServiceBusSessionProcessorOptions BuildSessionProcessorOptions(AzureServiceBusEndpoint endpoint) + { + var options = new ServiceBusSessionProcessorOptions + { + // Map the existing "parallel sessions" knob (ListenerCount, set via RequireSessions(count)) + // onto the processor's concurrency. A user may override this in ConfigureSessionProcessor. + MaxConcurrentSessions = endpoint.ListenerCount > 0 ? endpoint.ListenerCount : 1, + + // Preserve the in-session FIFO ordering the hand-rolled loop provided + MaxConcurrentCallsPerSession = 1 + }; + + endpoint.ConfigureSessionProcessor?.Invoke(options); + + // Reserved by Wolverine: the listener relies on the peek-lock model to explicitly complete, + // defer, and dead letter messages, so these cannot be honored from user configuration. + options.ReceiveMode = ServiceBusReceiveMode.PeekLock; + options.AutoCompleteMessages = false; + + return options; + } } \ No newline at end of file diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs index 01ad6cbf6..d8cf3fb3c 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs @@ -35,6 +35,19 @@ public AzureServiceBusEndpoint(AzureServiceBusTransport parent, Uri uri, Endpoin [IgnoreDescription] public AzureServiceBusTransport Parent { get; } + /// + /// Optional customization of the Azure Service Bus used + /// by session-enabled listeners for this endpoint. Setting this (directly, or the SessionIds + /// collection via RequireSessionsWithOnlyTheseIdentifiers(...)) switches the session listener away + /// from the default AcceptNextSession loop to a . Wolverine + /// reserves control of the properties it depends on for message acknowledgement (currently + /// ReceiveMode and AutoCompleteMessages), which are re-asserted after this action runs. This + /// is a multicast delegate so the SessionIds sugar and any explicit customization compose rather + /// than overwrite each other. + /// + [IgnoreDescription] + public Action? ConfigureSessionProcessor { get; set; } + /// /// The maximum number of messages to receive in a single batch when listening /// in either buffered or durable modes. The default is 20. diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEnvelope.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEnvelope.cs index ac52cf762..99444f9a5 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEnvelope.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEnvelope.cs @@ -17,6 +17,12 @@ public AzureServiceBusEnvelope(ProcessMessageEventArgs args) AzureMessage = args.Message; } + public AzureServiceBusEnvelope(ProcessSessionMessageEventArgs sessionArgs) + { + SessionArgs = sessionArgs; + AzureMessage = sessionArgs.Message; + } + public AzureServiceBusEnvelope(ServiceBusReceivedMessage message, ServiceBusReceiver sessionReceiver) { AzureMessage = message; @@ -31,6 +37,10 @@ public async Task CompleteAsync(CancellationToken token) { await Args.CompleteMessageAsync(AzureMessage, token); } + else if (SessionArgs != null) + { + await SessionArgs.CompleteMessageAsync(AzureMessage, token); + } else if (ServiceBusReceiver != null) { await ServiceBusReceiver.CompleteMessageAsync(AzureMessage, token); @@ -53,18 +63,22 @@ public async Task CompleteAsync(CancellationToken token) public Task DeferAsync(CancellationToken token) { - return Args?.DeferMessageAsync(AzureMessage, cancellationToken: token) ?? ServiceBusReceiver?.DeferMessageAsync(AzureMessage, cancellationToken: token) ?? + return Args?.DeferMessageAsync(AzureMessage, cancellationToken: token) + ?? SessionArgs?.DeferMessageAsync(AzureMessage, cancellationToken: token) + ?? ServiceBusReceiver?.DeferMessageAsync(AzureMessage, cancellationToken: token) ?? SessionReceiver?.DeferMessageAsync(AzureMessage, cancellationToken: token) ?? Task.CompletedTask; } public Task DeadLetterAsync(CancellationToken token, string? deadLetterReason = null, string? deadLetterErrorDescription = null) { return Args?.DeadLetterMessageAsync(AzureMessage, cancellationToken: token, deadLetterReason: deadLetterReason, deadLetterErrorDescription:deadLetterErrorDescription) + ?? SessionArgs?.DeadLetterMessageAsync(AzureMessage, cancellationToken: token, deadLetterReason: deadLetterReason, deadLetterErrorDescription:deadLetterErrorDescription) ?? ServiceBusReceiver?.DeadLetterMessageAsync(AzureMessage, cancellationToken: token, deadLetterReason: deadLetterReason, deadLetterErrorDescription:deadLetterErrorDescription) ?? SessionReceiver?.DeadLetterMessageAsync(AzureMessage, cancellationToken: token, deadLetterReason: deadLetterReason, deadLetterErrorDescription:deadLetterErrorDescription) ?? Task.CompletedTask; } private ProcessMessageEventArgs? Args { get; set; } + private ProcessSessionMessageEventArgs? SessionArgs { get; set; } private ServiceBusReceivedMessage AzureMessage { get; } private ServiceBusSessionReceiver? SessionReceiver { get; } diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusSessionListener.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusSessionListener.cs new file mode 100644 index 000000000..61459161e --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusSessionListener.cs @@ -0,0 +1,189 @@ +using Azure.Messaging.ServiceBus; +using JasperFx.Blocks; +using JasperFx.Core; +using JasperFx.Core.Reflection; +using Microsoft.Extensions.Logging; +using Wolverine.Runtime; +using Wolverine.Transports; +using Wolverine.Transports.Sending; + +namespace Wolverine.AzureServiceBus.Internal; + +/// +/// Session-enabled listener built on the Azure SDK's . This is +/// used in place of the hand-rolled AcceptNextSession loop (AzureServiceBusSessionListener) whenever the +/// endpoint has ConfigureSessionProcessor customization -- most notably when SessionIds is populated to pin +/// the listener to a fixed set of session identifiers (GH-3533). Mirrors InlineAzureServiceBusListener's +/// acknowledgement, dead lettering, and native scheduling. +/// +public class InlineAzureServiceBusSessionListener : IListener, ISupportDeadLetterQueue, ISupportNativeScheduling +{ + private readonly CancellationTokenSource _cancellation = new(); + private readonly RetryBlock _complete; + private readonly RetryBlock _deadLetter; + private readonly RetryBlock _defer; + private readonly AzureServiceBusEndpoint _endpoint; + private readonly ILogger _logger; + private readonly IIncomingMapper _mapper; + private readonly ServiceBusSessionProcessor _processor; + private readonly IReceiver _receiver; + private readonly ISender _requeue; + + public InlineAzureServiceBusSessionListener(AzureServiceBusEndpoint endpoint, + ILogger logger, + ServiceBusSessionProcessor processor, IReceiver receiver, + IIncomingMapper mapper, + ISender requeue) + { + _endpoint = endpoint; + _logger = logger; + _processor = processor; + _receiver = receiver; + _mapper = mapper; + _requeue = requeue; + + _complete = new RetryBlock((e, _) => { return e.CompleteAsync(_cancellation.Token); }, + _logger, _cancellation.Token); + + _defer = new RetryBlock(async (envelope, _) => + { + if (envelope is { } e) + { + await e.CompleteAsync(_cancellation.Token); + e.IsCompleted = true; + } + + await _requeue.SendAsync(envelope); + }, logger, _cancellation.Token); + + _deadLetter = + new RetryBlock( + (e, c) => e.DeadLetterAsync(_cancellation.Token, e.Exception?.GetType().NameInCode(), + e.Exception?.Message), logger, + _cancellation.Token); + + _processor.ProcessMessageAsync += processMessageAsync; + _processor.ProcessErrorAsync += processErrorAsync; + } + + public IHandlerPipeline? Pipeline => _receiver.Pipeline; + + public ValueTask CompleteAsync(Envelope envelope) + { + if (envelope is AzureServiceBusEnvelope e) + { + var task = _complete.PostAsync(e); + return new ValueTask(task); + } + + return ValueTask.CompletedTask; + } + + public ValueTask DeferAsync(Envelope envelope) + { + if (envelope is AzureServiceBusEnvelope e) + { + var task = _defer.PostAsync(e); + return new ValueTask(task); + } + + return ValueTask.CompletedTask; + } + + public async Task TryRequeueAsync(Envelope envelope) + { + if (envelope is AzureServiceBusEnvelope e) + { + await _defer.PostAsync(e); + return true; + } + + return false; + } + + public async ValueTask DisposeAsync() + { + _cancellation.Cancel(); + _complete.SafeDispose(); + _defer.SafeDispose(); + _deadLetter.SafeDispose(); + await _processor.DisposeAsync(); + } + + public Uri Address => _endpoint.Uri; + + public async ValueTask StopAsync() + { + await _processor.StopProcessingAsync(); + } + + public async Task MoveToErrorsAsync(Envelope envelope, Exception exception) + { + if (envelope is AzureServiceBusEnvelope e) + { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); + e.Exception = exception; + await _deadLetter.PostAsync(e); + } + } + + public bool NativeDeadLetterQueueEnabled => true; + + public async Task MoveToScheduledUntilAsync(Envelope envelope, DateTimeOffset time) + { + envelope.ScheduledTime = time; + await _requeue.SendAsync(envelope); + } + + public Task StartAsync() + { + return _processor.StartProcessingAsync(); + } + + private Task processErrorAsync(ProcessErrorEventArgs arg) + { + _logger.LogError(arg.Exception, "Error trying to receive Azure Service Bus message at {Uri}", _endpoint.Uri); + return Task.CompletedTask; + } + + private async Task processMessageAsync(ProcessSessionMessageEventArgs arg) + { + try + { + var envelope = new AzureServiceBusEnvelope(arg); + _mapper.MapIncomingToEnvelope(envelope, arg.Message); + + // If a ping, you're done, ack it and get out of there + if (envelope.IsPing()) + { + await CompleteAsync(envelope); + return; + } + + try + { + await _receiver.ReceivedAsync(this, envelope); + } + catch (Exception e) + { + _logger.LogError(e, "Failure to receive an incoming message with {Id}, trying to 'Defer' the message", + envelope.Id); + + try + { + await DeferAsync(envelope); + } + catch (Exception exception) + { + _logger.LogError(exception, "Failure trying to Nack a previously failed message {Id}", envelope.Id); + } + } + } + catch (Exception e) + { + await _deadLetter.PostAsync(new AzureServiceBusEnvelope(arg) { Exception = e }); + _logger.LogError(e, "Error while reading message {Id} from {Uri}", arg.Message.MessageId, + _endpoint.Uri); + } + } +} From 1bdb8a5af2775d73b8aca5f02536106df2001b35 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 20 Jul 2026 12:02:17 -0500 Subject: [PATCH 2/2] ci: re-trigger CI now that 5.0 PR workflows are enabled (#3548)