Stop Listener when disposed

Azure · Feb 10, 2022 · 2787f67 · 2787f67
1 parent f4b1757
commit 2787f67
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 8 deletions.
diff --git a/release_notes.md b/release_notes.md
@@ -5,5 +5,8 @@
 #### Version 4.2.0
 -  User configurable initial offset support [#79](https://github.com/Azure/azure-functions-eventhubs-extension/pull/79)
 
-**Release sprint:** Sprint 87
-[ [bugs](https://github.com/Azure/azure-functions-host/issues?q=is%3Aissue+milestone%3A%22Functions+Sprint+87%22+label%3Abug+is%3Aclosed) | [features](https://github.com/Azure/azure-functions-host/issues?q=is%3Aissue+milestone%3A%22Functions+Sprint+87%22+label%3Afeature+is%3Aclosed) ]
+#### Version 4.3.0
+- Adding explicit reference Microsoft.Azure.Amqp 2.4.11 [#99](https://github.com/Azure/azure-functions-eventhubs-extension/pull/99)
+
+#### Version 4.3.1
+- Stop Listener when disposed [#105](https://github.com/Azure/azure-functions-eventhubs-extension/pull/105)
diff --git a/src/Microsoft.Azure.WebJobs.Extensions.EventHubs/Listeners/EventHubListener.cs b/src/Microsoft.Azure.WebJobs.Extensions.EventHubs/Listeners/EventHubListener.cs
@@ -34,7 +34,8 @@ internal sealed class EventHubListener : IListener, IEventProcessorFactory, ISca
         private readonly bool _singleDispatch;
         private readonly EventHubOptions _options;
         private readonly ILogger _logger;
-        private bool _started;
+        private readonly SemaphoreSlim _stopSemaphoreSlim = new SemaphoreSlim(1, 1);
+        private bool _started;     
 
         private Lazy<EventHubsScaleMonitor> _scaleMonitor;
 
@@ -71,6 +72,7 @@ void IListener.Cancel()
 
         void IDisposable.Dispose()
         {
+            //StopAsync(CancellationToken.None).Wait();
         }
 
         public async Task StartAsync(CancellationToken cancellationToken)
@@ -81,11 +83,19 @@ public async Task StartAsync(CancellationToken cancellationToken)
 
         public async Task StopAsync(CancellationToken cancellationToken)
         {
-            if (_started)
+            await _stopSemaphoreSlim.WaitAsync();
+            try
             {
-                await _eventProcessorHost.UnregisterEventProcessorAsync();
+                if (_started)
+                {
+                    await _eventProcessorHost.UnregisterEventProcessorAsync();
+                }
+                _started = false;
+            }
+            finally
+            {
+                _stopSemaphoreSlim.Release();
             }
-            _started = false;
         }
 
         IEventProcessor IEventProcessorFactory.CreateEventProcessor(PartitionContext context)
@@ -132,7 +142,7 @@ public Task CloseAsync(PartitionContext context, CloseReason reason)
             {
                 // signal cancellation for any in progress executions 
                 _cts.Cancel();
-                
+
                 _logger.LogDebug(GetOperationDetails(context, $"CloseAsync, {reason.ToString()}"));
                 return Task.CompletedTask;
             }

diff --git a/src/Microsoft.Azure.WebJobs.Extensions.EventHubs/WebJobs.Extensions.EventHubs.csproj b/src/Microsoft.Azure.WebJobs.Extensions.EventHubs/WebJobs.Extensions.EventHubs.csproj
@@ -6,7 +6,7 @@
     <RootNamespace>Microsoft.Azure.WebJobs.EventHubs</RootNamespace>
     <PackageId>Microsoft.Azure.WebJobs.Extensions.EventHubs</PackageId>
     <Description>Microsoft Azure WebJobs SDK EventHubs Extension</Description>
-    <Version>4.3.0</Version>
+    <Version>4.3.1</Version>
     <CommitHash Condition="$(CommitHash) == ''">N/A</CommitHash>
     <InformationalVersion>$(Version) Commit hash: $(CommitHash)</InformationalVersion>
     <Authors>Microsoft</Authors>