Stop Listener when disposed (#104)

Azure · Feb 11, 2022 · 3a6f3ee · 3a6f3ee
1 parent f4b1757
commit 3a6f3ee
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 8 deletions.
diff --git a/release_notes.md b/release_notes.md
@@ -5,5 +5,9 @@
 #### 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)
+- Add listener details [#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,6 +34,8 @@ internal sealed class EventHubListener : IListener, IEventProcessorFactory, ISca
         private readonly bool _singleDispatch;
         private readonly EventHubOptions _options;
         private readonly ILogger _logger;
+        private readonly SemaphoreSlim _stopSemaphoreSlim = new SemaphoreSlim(1, 1);
+        private readonly string _details;
         private bool _started;
 
         private Lazy<EventHubsScaleMonitor> _scaleMonitor;
@@ -62,6 +64,8 @@ public EventHubListener(
             _options = options;
             _logger = logger;
             _scaleMonitor = new Lazy<EventHubsScaleMonitor>(() => new EventHubsScaleMonitor(_functionId, _eventHubName, _consumerGroup, _connectionString, _storageConnectionString, _logger, blobContainer));
+            _details = $"'namespace='{eventProcessorHost?.EndpointAddress}', eventHub='{eventProcessorHost?.EventHubPath}', " +
+                $"consumerGroup='{eventProcessorHost?.ConsumerGroupName}', functionId='{functionId}', singleDispatch='{singleDispatch}'";
         }
 
         void IListener.Cancel()
@@ -71,21 +75,37 @@ void IListener.Cancel()
 
         void IDisposable.Dispose()
         {
+            StopAsync(CancellationToken.None).Wait();
         }
 
         public async Task StartAsync(CancellationToken cancellationToken)
         {
             await _eventProcessorHost.RegisterEventProcessorFactoryAsync(this, _options.EventProcessorOptions);
             _started = true;
+
+            _logger.LogDebug($"EventHub listener started ({_details})");
         }
 
         public async Task StopAsync(CancellationToken cancellationToken)
         {
-            if (_started)
+            await _stopSemaphoreSlim.WaitAsync();
+            try
+            {
+                if (_started)
+                {
+                    await _eventProcessorHost.UnregisterEventProcessorAsync();
+                    _logger.LogDebug($"EventHub listener stopped ({_details})");
+                }
+                else
+                {
+                    _logger.LogDebug($"EventHub listener is already stopped ({_details})");
+                }
+                _started = false;
+            }
+            finally
             {
-                await _eventProcessorHost.UnregisterEventProcessorAsync();
+                _stopSemaphoreSlim.Release();
             }
-            _started = false;
         }
 
         IEventProcessor IEventProcessorFactory.CreateEventProcessor(PartitionContext context)
@@ -132,7 +152,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>

diff --git a/test/Microsoft.Azure.WebJobs.Extensions.EventHubs.Tests/EventHubListenerTests.cs b/test/Microsoft.Azure.WebJobs.Extensions.EventHubs.Tests/EventHubListenerTests.cs
@@ -11,6 +11,7 @@
 using Microsoft.Azure.EventHubs.Processor;
 using Microsoft.Azure.WebJobs.EventHubs.Listeners;
 using Microsoft.Azure.WebJobs.Host.Executors;
+using Microsoft.Azure.WebJobs.Host.Listeners;
 using Microsoft.Azure.WebJobs.Host.Scale;
 using Microsoft.Azure.WebJobs.Host.TestCommon;
 using Microsoft.Extensions.Logging;
@@ -198,7 +199,6 @@ public void GetMonitor_ReturnsExpectedValue()
             var functionId = "FunctionId";
             var eventHubName = "EventHubName";
             var consumerGroup = "ConsumerGroup";
-            var storageUri = new Uri("https://eventhubsteststorageaccount.blob.core.windows.net/");
             var testLogger = new TestLogger("Test");
             var listener = new EventHubListener(
                                     functionId,
@@ -222,5 +222,30 @@ public void GetMonitor_ReturnsExpectedValue()
 
             Assert.Same(scaleMonitor, scaleMonitor2);
         }
+
+        [Fact]
+        public void Dispose_Calls_StopAsync()
+        {                       
+            string functionId = "FunctionId";
+            string eventHubName = "EventHubName";
+            string consumerGroup = "ConsumerGroup";
+            var storageUri = new Uri("https://eventhubsteststorageaccount.blob.core.windows.net/");
+            var testLogger = new TestLogger("Test");
+            EventHubListener listener = new EventHubListener(
+                                    functionId,
+                                    eventHubName,
+                                    consumerGroup,
+                                    "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=abc123=",
+                                    "DefaultEndpointsProtocol=https;AccountName=EventHubScaleMonitorFakeTestAccount;AccountKey=ABCDEFG;EndpointSuffix=core.windows.net",
+                                    new Mock<ITriggeredFunctionExecutor>(MockBehavior.Strict).Object,
+                                    null,
+                                    false,
+                                    new EventHubOptions(),
+                                    testLogger,
+                                    new Mock<CloudBlobContainer>(MockBehavior.Strict, new Uri("https://eventhubsteststorageaccount.blob.core.windows.net/azure-webjobs-eventhub")).Object);
+
+            (listener as IDisposable).Dispose();
+            Assert.Single(testLogger.GetLogMessages().Where(x => x.FormattedMessage.StartsWith("EventHub listener is already stopped (")));
+        }
     }
 }