Fixes ServiceBusProcessor.DisposeAsync exceptions (Azure#18027)

This fixes an issues when a ServiceBusProcessor is disposed after it has been closed or is disposed multiple times. If a processor is used within a using block and also has Close explicitly called before the processor goes out of scope, it will throw an NRE.

Regarding multiple calls to dispose, I can't think of a good use case for why somebody would do that but it is documented behavior for IAsyncDisposable.

https://docs.microsoft.com/en-us/dotnet/api/system.iasyncdisposable.disposeasync?view=dotnet-plat-ext-5.0#remarks
> If an object's DisposeAsync method is called more than once, the object must ignore all calls after the first one and synchronously return a successfully completed ValueTask. The object must not throw an exception if its DisposeAsync method is called multiple times.

Author: aarondandy

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ReceiverManager.cs

@@ -72,13 +72,17 @@ public virtual async Task CloseReceiverIfNeeded(
             CancellationToken cancellationToken,
             bool forceClose = false)
         {
-            try
+            var capturedReceiver = Receiver;
+            if (capturedReceiver != null)
             {
-                await Receiver.DisposeAsync().ConfigureAwait(false);
-            }
-            finally
-            {
-                Receiver = null;
+                try
+                {
+                    await capturedReceiver.DisposeAsync().ConfigureAwait(false);
+                }
+                finally
+                {
+                    Receiver = null;
+                }
             }
         }
 

sdk/servicebus/Azure.Messaging.ServiceBus/tests/Infrastructure/ServiceBusTestBase.cs

@@ -4,7 +4,9 @@
 using System;
 using System.Collections.Generic;
 using System.Security.Cryptography;
+using System.Threading;
 using System.Threading.Tasks;
+using Azure.Messaging.ServiceBus.Core;
 using Moq;
 using NUnit.Framework;
 
@@ -79,11 +81,31 @@ protected byte[] GetRandomBuffer(long size)
 
         internal ServiceBusConnection GetMockedConnection()
         {
+            var mockTransportReceiver = new Mock<TransportReceiver>();
+            mockTransportReceiver
+                .Setup(receiver => receiver.ReceiveMessagesAsync(It.IsAny<int>(), It.IsAny<TimeSpan?>(), It.IsAny<CancellationToken>()))
+                .Returns(async (int maximumMessageCount, TimeSpan? maxWaitTime, CancellationToken cancellationToken) =>
+                {
+                    await Task.Delay(Timeout.Infinite, cancellationToken);
+                    throw new NotImplementedException();
+                });
+
             var mockConnection = new Mock<ServiceBusConnection>();
 
             mockConnection
                 .Setup(connection => connection.RetryOptions)
                 .Returns(new ServiceBusRetryOptions());
+
+            mockConnection
+                .Setup(connection => connection.CreateTransportReceiver(
+                    It.IsAny<string>(),
+                    It.IsAny<ServiceBusRetryPolicy>(),
+                    It.IsAny<ServiceBusReceiveMode>(),
+                    It.IsAny<uint>(),
+                    It.IsAny<string>(),
+                    It.IsAny<string>(),
+                    It.IsAny<bool>()))
+                .Returns(mockTransportReceiver.Object);
             return mockConnection.Object;
         }
     }

sdk/servicebus/Azure.Messaging.ServiceBus/tests/Processor/ProcessorTests.cs

@@ -281,5 +281,40 @@ public void UserSettledPropertySetCorrectlyOnException()
                 Throws.InstanceOf<Exception>());
             Assert.IsFalse(msg.IsSettled);
         }
+
+        [Test]
+        public async Task CanDisposeStartedProcessorMultipleTimes()
+        {
+            var processor = new ServiceBusProcessor(
+                GetMockedConnection(),
+                "entityPath",
+                false,
+                new ServiceBusPlugin[] { },
+                new ServiceBusProcessorOptions());
+            processor.ProcessMessageAsync += _ => Task.CompletedTask;
+            processor.ProcessErrorAsync += _ => Task.CompletedTask;
+            await processor.StartProcessingAsync().ConfigureAwait(false);
+
+            await processor.DisposeAsync();
+            await processor.DisposeAsync();
+        }
+
+        [Test]
+        public async Task CanDisposeClosedProcessor()
+        {
+            var processor = new ServiceBusProcessor(
+                GetMockedConnection(),
+                "entityPath",
+                false,
+                new ServiceBusPlugin[] { },
+                new ServiceBusProcessorOptions());
+
+            processor.ProcessMessageAsync += _ => Task.CompletedTask;
+            processor.ProcessErrorAsync += _ => Task.CompletedTask;
+            await processor.StartProcessingAsync().ConfigureAwait(false);
+            await processor.CloseAsync();
+
+            await processor.DisposeAsync();
+        }
     }
 }