Graceful shutdown (message draining) (#102)

Author: sidkri

src/Microsoft.Azure.WebJobs.Extensions.ServiceBus/Listeners/ServiceBusListener.cs

@@ -38,6 +38,8 @@ internal sealed class ServiceBusListener : IListener, IScaleMonitorProvider
         private ClientEntity _clientEntity;
         private bool _disposed;
         private bool _started;
+        // Serialize execution of StopAsync to avoid calling Unregister* concurrently
+        private readonly SemaphoreSlim _stopAsyncSemaphore = new SemaphoreSlim(1, 1);
 
         private IMessageSession _messageSession;
         private SessionMessageProcessor _sessionMessageProcessor;
@@ -118,27 +120,53 @@ public Task StartAsync(CancellationToken cancellationToken)
         public async Task StopAsync(CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
-
-            if (!_started)
+            await _stopAsyncSemaphore.WaitAsync();
             {
-                throw new InvalidOperationException("The listener has not yet been started or has already been stopped.");
-            }
+                try
+                {
+                    if (!_started)
+                    {
+                        throw new InvalidOperationException("The listener has not yet been started or has already been stopped.");
+                    }
 
-            // cancel our token source to signal any in progress
-            // ProcessMessageAsync invocations to cancel
-            _cancellationTokenSource.Cancel();
+                    // Unregister* methods stop new messages from being processed while allowing in-flight messages to complete.
+                    // As the amount of time functions are allowed to complete processing varies by SKU, we specify max timespan
+                    // as the amount of time Service Bus SDK should wait for in-flight messages to complete procesing after 
+                    // unregistering the message handler so that functions have as long as the host continues to run time to complete.
+                    if (_singleDispatch)
+                    {
+                        if (_isSessionsEnabled)
+                        {
+                            if (_clientEntity != null)
+                            {
+                                if (_clientEntity is QueueClient queueClient)
+                                {
+                                    await queueClient.UnregisterSessionHandlerAsync(TimeSpan.MaxValue);
+                                }
+                                else
+                                {
+                                    SubscriptionClient subscriptionClient = _clientEntity as SubscriptionClient;
+                                    await subscriptionClient.UnregisterSessionHandlerAsync(TimeSpan.MaxValue);
+                                }
+                            }
+                        }
+                        else
+                        {
+                            if (_receiver != null && _receiver.IsValueCreated)
+                            {
+                                await Receiver.UnregisterMessageHandlerAsync(TimeSpan.MaxValue);
+                            }
+                        }
+                    }
+                    // Batch processing will be stopped via the _started flag on its next iteration
 
-            if (_receiver != null && _receiver.IsValueCreated)
-            {
-                await Receiver.CloseAsync();
-                _receiver = CreateMessageReceiver();
-            }
-            if (_clientEntity != null)
-            {
-                await _clientEntity.CloseAsync();
-                _clientEntity = null;
+                    _started = false;
+                }
+                finally
+                {
+                    _stopAsyncSemaphore.Release();
+                }
             }
-            _started = false;
         }
 
         public void Cancel()
@@ -176,6 +204,9 @@ public void Dispose()
                     _clientEntity = null;
                 }
 
+                _stopAsyncSemaphore.Dispose();
+                _cancellationTokenSource.Dispose();
+
                 _disposed = true;
             }
         }
@@ -192,37 +223,39 @@ private Lazy<SessionClient> CreateSessionClient()
 
         internal async Task ProcessMessageAsync(Message message, CancellationToken cancellationToken)
         {
-            CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationTokenSource.Token);
-
-            if (!await _messageProcessor.BeginProcessingMessageAsync(message, linkedCts.Token))
+            using (CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationTokenSource.Token))
             {
-                return;
-            }
+                if (!await _messageProcessor.BeginProcessingMessageAsync(message, linkedCts.Token))
+                {
+                    return;
+                }
 
-            ServiceBusTriggerInput input = ServiceBusTriggerInput.CreateSingle(message);
-            input.MessageReceiver = Receiver;
+                ServiceBusTriggerInput input = ServiceBusTriggerInput.CreateSingle(message);
+                input.MessageReceiver = Receiver;
 
-            TriggeredFunctionData data = input.GetTriggerFunctionData();
-            FunctionResult result = await _triggerExecutor.TryExecuteAsync(data, linkedCts.Token);
-            await _messageProcessor.CompleteProcessingMessageAsync(message, result, linkedCts.Token);
+                TriggeredFunctionData data = input.GetTriggerFunctionData();
+                FunctionResult result = await _triggerExecutor.TryExecuteAsync(data, linkedCts.Token);
+                await _messageProcessor.CompleteProcessingMessageAsync(message, result, linkedCts.Token);
+            }
         }
 
         internal async Task ProcessSessionMessageAsync(IMessageSession session, Message message, CancellationToken cancellationToken)
         {
-            CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationTokenSource.Token);
-
-            _messageSession = session;
-            if (!await _sessionMessageProcessor.BeginProcessingMessageAsync(session, message, linkedCts.Token))
+            using (CancellationTokenSource linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationTokenSource.Token))
             {
-                return;
-            }
+                _messageSession = session;
+                if (!await _sessionMessageProcessor.BeginProcessingMessageAsync(session, message, linkedCts.Token))
+                {
+                    return;
+                }
 
-            ServiceBusTriggerInput input = ServiceBusTriggerInput.CreateSingle(message);
-            input.MessageReceiver = session;
+                ServiceBusTriggerInput input = ServiceBusTriggerInput.CreateSingle(message);
+                input.MessageReceiver = session;
 
-            TriggeredFunctionData data = input.GetTriggerFunctionData();
-            FunctionResult result = await _triggerExecutor.TryExecuteAsync(data, linkedCts.Token);
-            await _sessionMessageProcessor.CompleteProcessingMessageAsync(session, message, result, linkedCts.Token);
+                TriggeredFunctionData data = input.GetTriggerFunctionData();
+                FunctionResult result = await _triggerExecutor.TryExecuteAsync(data, linkedCts.Token);
+                await _sessionMessageProcessor.CompleteProcessingMessageAsync(session, message, result, linkedCts.Token);
+            }
         }
 
         internal void StartMessageBatchReceiver(CancellationToken cancellationToken)
@@ -246,6 +279,7 @@ internal void StartMessageBatchReceiver(CancellationToken cancellationToken)
                     {
                         if (!_started || cancellationToken.IsCancellationRequested)
                         {
+                            _logger.LogInformation("Message processing has been stopped or cancelled");
                             return;
                         }
 

src/Microsoft.Azure.WebJobs.Extensions.ServiceBus/WebJobs.Extensions.ServiceBus.csproj

@@ -6,7 +6,7 @@
     <RootNamespace>Microsoft.Azure.WebJobs.ServiceBus</RootNamespace>
     <PackageId>Microsoft.Azure.WebJobs.Extensions.ServiceBus</PackageId>
     <Description>Microsoft Azure WebJobs SDK ServiceBus Extension</Description>
-    <Version>4.1.2</Version>
+    <Version>4.2.0$(VersionSuffix)</Version>
     <CommitHash Condition="$(CommitHash) == ''">N/A</CommitHash>
     <InformationalVersion>$(Version) Commit hash: $(CommitHash)</InformationalVersion>
     <Authors>Microsoft</Authors>
@@ -28,6 +28,7 @@
     <StyleCopTreatErrorsAsWarnings>false</StyleCopTreatErrorsAsWarnings>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <WarningsAsErrors />
+    <NoWarn>1701;1702</NoWarn>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
@@ -37,16 +38,17 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Azure.ServiceBus" Version="4.1.1" />
-    <PackageReference Include="Microsoft.Azure.WebJobs" Version="3.0.16" />
-    <PackageReference Include="Microsoft.Azure.WebJobs.Sources" Version="3.0.16" />
+    <PackageReference Include="Microsoft.Azure.ServiceBus" Version="4.2.0" />
+    <PackageReference Include="Microsoft.Azure.WebJobs" Version="3.0.19" />
+    <PackageReference Include="Microsoft.Azure.WebJobs.Sources" Version="3.0.19" />
     <PackageReference Include="StyleCop.Analyzers" Version="1.1.0-beta004">
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>
   </ItemGroup>
 
   <ItemGroup>
     <AdditionalFiles Include="..\..\stylecop.json" Link="stylecop.json" />
+    <None Include="..\..\webjobs.png" Pack="true" PackagePath="\" />
   </ItemGroup>
 
 </Project>

test/Microsoft.Azure.WebJobs.Extensions.ServiceBus.Tests/README.md

@@ -0,0 +1,34 @@
+# Service Bus Extension for Azure Functions guide to running integration tests locally
+Integration tests are implemented in the `EndToEndTests` and `SessionsEndToEndTests` classes and require special configuration to execute locally in Visual Studio or via dotnet test.  
+
+All configuration is done via a json file called `appsettings.tests` which on windows should be located in the `%USERPROFILE%\.azurefunctions` folder (e.g. `C:\Users\user123\.azurefunctions`)
+
+**Note:** *The specifics of the configuration will change when the validation code is modified so check the code for the latest configuration if the tests do not pass as this readme file may not have been updated with each code change.*
+
+Create the appropriate Azure resources if needed as explained below and create or update the `appsettings.tests` file in the location specified above by copying the configuration below and replacing all the `PLACEHOLDER` values
+
+appsettings.tests contents
+```
+{
+    "ConnectionStrings": {
+        "ServiceBus": "PLACEHOLDER",
+        "ServiceBusSecondary": "PLACEHOLDER"
+    },
+    "AzureWebJobsStorage": "PLACEHOLDER"
+}
+```
+Create a storage account and configure its connection string into `AzureWebJobsStorage`.  This will be used by the webjobs hosts created by the tests.
+
+Create two service bus namespaces and configure their connection strings in `ConnectionStrings:ServiceBus` and `ConnectionStrings:ServiceBusSecondary`.  
+1. In the namespace configured into `ConnectionStrings:ServiceBus`, create queues with the following names:
+    1. `core-test-queue1`
+    2. `core-test-queue2`
+    3. `core-test-queue3`
+    4. `core-test-queue1-sessions` (enable sessions when creating)
+2. In the namespace configured into `ConnectionStrings:ServiceBus`, create topics and subscriptions with the following names:
+    1. `core-test-topic1` with two subscriptions: `sub1` and `sub2`
+    2. `core-test-topic1-sessions` with one subscription: `sub1-sessions` (enable sessions in the subscription when creating)
+2. In the namespace configured into `ConnectionStrings:ServiceBusSecondary`, create queues with the following names:
+    1. `core-test-queue1`
+
+  Change the message lock duration setting on all queues and subscriptions to 5 minutes to all for delays associated with stepping through code in debug mode.