diff --git a/docs/guide/messaging/transports/azureservicebus/listening.md b/docs/guide/messaging/transports/azureservicebus/listening.md index 65f712f19..63c2d3456 100644 --- a/docs/guide/messaging/transports/azureservicebus/listening.md +++ b/docs/guide/messaging/transports/azureservicebus/listening.md @@ -45,6 +45,62 @@ await host.StartAsync(); snippet source | anchor +## Configuring the ServiceBusProcessor + +When a listener runs [inline](/guide/messaging/endpoints.html#endpoint-types) (`ProcessInline()`), Wolverine +creates an Azure Service Bus `ServiceBusProcessor` for the queue or subscription. You can customize the +underlying `ServiceBusProcessorOptions` with `ConfigureProcessor()` on either a queue or a subscription +listener. + +The most common reason to reach for this is a long-running inline handler. The Azure SDK only renews a +message's lock for five minutes by default (`MaxAutoLockRenewalDuration`). Any inline handler that runs +longer than that loses the lock, so the completion acknowledgement silently fails and Azure Service Bus +redelivers the message — resulting in duplicate work until the message is finally dead-lettered. Raising +`MaxAutoLockRenewalDuration` keeps the lock alive for the length of your handler: + + + +```cs +var builder = Host.CreateApplicationBuilder(); +builder.UseWolverine(opts => +{ + // One way or another, you're probably pulling the Azure Service Bus + // connection string out of configuration + var azureServiceBusConnectionString = builder + .Configuration + .GetConnectionString("azure-service-bus")!; + + opts.UseAzureServiceBus(azureServiceBusConnectionString).AutoProvision(); + + opts.ListenToAzureServiceBusQueue("incoming") + + // Inline listeners create an Azure Service Bus ServiceBusProcessor. By default the + // Azure SDK only renews the message lock for five minutes, so an inline handler that + // runs longer than that loses its lock and the message is redelivered. Raise the + // renewal window here so long-running inline handlers keep their lock. + .ConfigureProcessor(processorOptions => + { + processorOptions.MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(30); + }) + + // Run the handler inline against the ServiceBusProcessor + .ProcessInline(); +}); + +using var host = builder.Build(); +await host.StartAsync(); +``` +snippet source | anchor + + +::: warning +Wolverine's inline listener acknowledges, defers, and dead letters messages against the message lock, so it +requires the peek-lock receive model. The `ReceiveMode` you set in `ConfigureProcessor()` is therefore +ignored and always re-asserted to `ServiceBusReceiveMode.PeekLock`. All other `ServiceBusProcessorOptions` +properties are honored. Session-based listeners (`RequireSessions()`) do not use a `ServiceBusProcessor` and +are not affected by this option. +::: + ## Conventional Listener Configuration In the case of listening to a large number of queues, it may be beneficial diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/DocumentationSamples.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/DocumentationSamples.cs index 659afdb61..b5e013580 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/DocumentationSamples.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/DocumentationSamples.cs @@ -119,6 +119,41 @@ public async Task configuring_a_listener() #endregion } + public async Task configuring_processor_options() + { + #region sample_configuring_azure_service_bus_processor_options + var builder = Host.CreateApplicationBuilder(); + builder.UseWolverine(opts => + { + // One way or another, you're probably pulling the Azure Service Bus + // connection string out of configuration + var azureServiceBusConnectionString = builder + .Configuration + .GetConnectionString("azure-service-bus")!; + + opts.UseAzureServiceBus(azureServiceBusConnectionString).AutoProvision(); + + opts.ListenToAzureServiceBusQueue("incoming") + + // Inline listeners create an Azure Service Bus ServiceBusProcessor. By default the + // Azure SDK only renews the message lock for five minutes, so an inline handler that + // runs longer than that loses its lock and the message is redelivered. Raise the + // renewal window here so long-running inline handlers keep their lock. + .ConfigureProcessor(processorOptions => + { + processorOptions.MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(30); + }) + + // Run the handler inline against the ServiceBusProcessor + .ProcessInline(); + }); + + using var host = builder.Build(); + await host.StartAsync(); + + #endregion + } + public async Task configure_buffered_listener() { var builder = Host.CreateApplicationBuilder(); diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_processor_options.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_processor_options.cs new file mode 100644 index 000000000..b4f6dd6ce --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_processor_options.cs @@ -0,0 +1,85 @@ +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_processor_options +{ + [Fact] + public void configure_processor_action_is_stored_on_the_queue_endpoint() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var configuration = new AzureServiceBusQueueListenerConfiguration(queue); + configuration.ConfigureProcessor(o => o.MaxAutoLockRenewalDuration = 30.Minutes()); + + // Apply the delayed configuration as Wolverine would at bootstrapping time + ((IDelayedEndpointConfiguration)configuration).Apply(); + + queue.ConfigureProcessor.ShouldNotBeNull(); + } + + [Fact] + public void configure_processor_action_is_stored_on_the_subscription_endpoint() + { + var transport = new AzureServiceBusTransport(); + var topic = transport.Topics["topic1"]; + var subscription = topic.FindOrCreateSubscription("sub1"); + + var configuration = new AzureServiceBusSubscriptionListenerConfiguration(subscription); + configuration.ConfigureProcessor(o => o.MaxAutoLockRenewalDuration = 30.Minutes()); + + ((IDelayedEndpointConfiguration)configuration).Apply(); + + subscription.ConfigureProcessor.ShouldNotBeNull(); + } + + [Fact] + public void build_processor_options_applies_the_configured_action() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + queue.ConfigureProcessor = o => + { + o.MaxAutoLockRenewalDuration = 30.Minutes(); + o.PrefetchCount = 12; + }; + + var options = AzureServiceBusTransport.BuildProcessorOptions(queue); + + options.MaxAutoLockRenewalDuration.ShouldBe(30.Minutes()); + options.PrefetchCount.ShouldBe(12); + } + + [Fact] + public void build_processor_options_reasserts_peek_lock_receive_mode() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + // A user trying to switch to ReceiveAndDelete would break Wolverine's + // inline acknowledgement, so the transport must re-assert PeekLock. + queue.ConfigureProcessor = o => o.ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete; + + var options = AzureServiceBusTransport.BuildProcessorOptions(queue); + + options.ReceiveMode.ShouldBe(ServiceBusReceiveMode.PeekLock); + } + + [Fact] + public void build_processor_options_is_valid_when_no_action_is_configured() + { + var transport = new AzureServiceBusTransport(); + var queue = transport.Queues["incoming"]; + + var options = AzureServiceBusTransport.BuildProcessorOptions(queue); + + options.ShouldNotBeNull(); + options.ReceiveMode.ShouldBe(ServiceBusReceiveMode.PeekLock); + } +} diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs index 580cc5394..e82e59a3a 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; @@ -64,6 +65,22 @@ public AzureServiceBusQueueListenerConfiguration ConfigureQueue(Action + /// Customize the Azure Service Bus used by this + /// listener when running in the inline (ProcessInline()) mode. This is the way to raise + /// for inline handlers that + /// run longer than the Azure SDK's default of five minutes. Wolverine reserves control of the + /// properties it depends on for message acknowledgement (currently ReceiveMode), which are + /// re-asserted after this action runs. + /// + /// + /// + public AzureServiceBusQueueListenerConfiguration ConfigureProcessor(Action configure) + { + add(e => e.ConfigureProcessor = configure); + return this; + } + /// /// 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/AzureServiceBusSubscriptionListenerConfiguration.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs index 94e1eaedc..1cd40a1f5 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; @@ -63,6 +64,22 @@ public AzureServiceBusSubscriptionListenerConfiguration ConfigureSubscription(Ac return this; } + /// + /// Customize the Azure Service Bus used by this + /// subscription listener when running in the inline (ProcessInline()) mode. This is the way + /// to raise for inline handlers + /// that run longer than the Azure SDK's default of five minutes. Wolverine reserves control of the + /// properties it depends on for message acknowledgement (currently ReceiveMode), which are + /// re-asserted after this action runs. + /// + /// + /// + public AzureServiceBusSubscriptionListenerConfiguration ConfigureProcessor(Action configure) + { + add(e => e.ConfigureProcessor = configure); + return this; + } + /// /// Configure the underlying Azure Service Bus Subscription rule. This is only applicable when /// Wolverine is creating the Subscription. diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs index 21c3b059e..9edc3374e 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs @@ -66,7 +66,7 @@ private async Task buildListenerForQueue(IWolverineRuntime runtime, I if (queue.Mode == EndpointMode.Inline) { - var messageProcessor = BusClient.CreateProcessor(queue.QueueName); + var messageProcessor = BusClient.CreateProcessor(queue.QueueName, BuildProcessorOptions(queue)); var inlineListener = new InlineAzureServiceBusListener(queue, runtime.LoggerFactory.CreateLogger(), messageProcessor, receiver, @@ -122,7 +122,7 @@ private async Task buildListenerForSubscription(IWolverineRuntime run if (subscription.Mode == EndpointMode.Inline) { - var messageProcessor = BusClient.CreateProcessor(subscription.Topic.TopicName, subscription.SubscriptionName); + var messageProcessor = BusClient.CreateProcessor(subscription.Topic.TopicName, subscription.SubscriptionName, BuildProcessorOptions(subscription)); var inlineListener = new InlineAzureServiceBusListener(subscription, runtime.LoggerFactory.CreateLogger(), messageProcessor, receiver, mapper, requeue ); @@ -138,4 +138,23 @@ private async Task buildListenerForSubscription(IWolverineRuntime run return listener; } + + // Builds the ServiceBusProcessorOptions for an inline listener, applying any user supplied + // customization and then re-asserting the properties that Wolverine's inline acknowledgement + // logic depends on. InlineAzureServiceBusListener explicitly completes, defers, and dead + // letters messages against the message lock, so the processor must run in PeekLock mode; the + // ReceiveAndDelete mode would remove messages before Wolverine can process them and would break + // dead lettering and deferral. + internal static ServiceBusProcessorOptions BuildProcessorOptions(AzureServiceBusEndpoint endpoint) + { + var options = new ServiceBusProcessorOptions(); + + endpoint.ConfigureProcessor?.Invoke(options); + + // Reserved by Wolverine: the inline listener relies on the peek-lock model to complete, + // defer, and dead letter messages, so this cannot be honored from user configuration. + options.ReceiveMode = ServiceBusReceiveMode.PeekLock; + + 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..8e25daf69 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs @@ -49,6 +49,15 @@ public AzureServiceBusEndpoint(AzureServiceBusTransport parent, Uri uri, Endpoin /// public TimeSpan MaximumWaitTime { get; set; } = 5.Seconds(); + /// + /// Optional customization of the Azure Service Bus used + /// by inline listeners for this endpoint. Wolverine reserves control of the properties it depends + /// on for correct message acknowledgement (see AzureServiceBusTransport.Listening), so those will + /// be re-asserted after this action runs. + /// + [IgnoreDescription] + public Action? ConfigureProcessor { get; set; } + public abstract ValueTask CheckAsync(); public abstract ValueTask TeardownAsync(ILogger logger); public abstract ValueTask SetupAsync(ILogger logger);