Enable a way to Unregister Message Handler and Session Handler

sdk/servicebus/Microsoft.Azure.ServiceBus/src/Core/IReceiverClient.cs

@@ -63,6 +63,11 @@ public interface IReceiverClient : IClientEntity
         /// <remarks>Enable prefetch to speed up the receive rate.</remarks>
         void RegisterMessageHandler(Func<Message, CancellationToken, Task> handler, MessageHandlerOptions messageHandlerOptions);
 
+        /// <summary>
+        /// Unregister message handler from the receiver if there is an active message handler registered. 
+        /// </summary>
+        Task UnregisterMessageHandler();
+
         /// <summary>
         /// Completes a <see cref="Message"/> using its lock token. This will delete the message from the queue.
         /// </summary>
@@ -115,4 +120,4 @@ public interface IReceiverClient : IClientEntity
         /// </remarks>
         Task DeadLetterAsync(string lockToken, string deadLetterReason, string deadLetterErrorDescription = null);
     }
-}
+}

sdk/servicebus/Microsoft.Azure.ServiceBus/src/Core/MessageReceiver.cs

@@ -899,6 +899,38 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
             this.OnMessageHandler(messageHandlerOptions, handler);
         }
 
+        /// <summary>
+        /// Unregister message hander from the receiver if there is an active message handler registered. 
+        /// </summary>
+        public async Task UnregisterMessageHandler()
+        {
+            this.ThrowIfClosed();
+
+            MessagingEventSource.Log.UnregisterMessageHandlerStart(this.ClientId);
+            lock (this.messageReceivePumpSyncLock)
+            {
+                if (this.receivePump == null || this.receivePumpCancellationTokenSource.IsCancellationRequested)
+                {
+                    return;
+                }
+
+                this.receivePumpCancellationTokenSource.Cancel();
+                this.receivePumpCancellationTokenSource.Dispose();
+            }
+
+            while (this.receivePump != null
+                && this.receivePump.maxConcurrentCallsSemaphoreSlim.CurrentCount < this.receivePump.registerHandlerOptions.MaxConcurrentCalls)
+            {
+                await Task.Delay(10).ConfigureAwait(false);
+            }
+
+            lock (this.messageReceivePumpSyncLock)
+            {
+                this.receivePump = null;
+            }
+            MessagingEventSource.Log.UnregisterMessageHandlerStop(this.ClientId);
+        }
+
         /// <summary>
         /// Registers a <see cref="ServiceBusPlugin"/> to be used with this receiver.
         /// </summary>

sdk/servicebus/Microsoft.Azure.ServiceBus/src/MessageReceivePump.cs

@@ -12,14 +12,15 @@ namespace Microsoft.Azure.ServiceBus
 
     sealed class MessageReceivePump
     {
+        public readonly SemaphoreSlim maxConcurrentCallsSemaphoreSlim;
+        public readonly MessageHandlerOptions registerHandlerOptions;
         readonly Func<Message, CancellationToken, Task> onMessageCallback;
         readonly string endpoint;
-        readonly MessageHandlerOptions registerHandlerOptions;
         readonly IMessageReceiver messageReceiver;
         readonly CancellationToken pumpCancellationToken;
-        readonly SemaphoreSlim maxConcurrentCallsSemaphoreSlim;
         readonly ServiceBusDiagnosticSource diagnosticSource;
 
+
         public MessageReceivePump(IMessageReceiver messageReceiver,
             MessageHandlerOptions registerHandlerOptions,
             Func<Message, CancellationToken, Task> callback,

sdk/servicebus/Microsoft.Azure.ServiceBus/src/MessagingEventSource.cs

@@ -1380,6 +1380,24 @@ public void ManagementSerializationException(string objectName, string details =
                 this.WriteEvent(117, objectName, details);
             }
         }
+
+        [Event(118, Level = EventLevel.Informational, Message = "{0}: Unregister MessageHandler start.")]
+        public void UnregisterMessageHandlerStart(string clientId)
+        {
+            if (this.IsEnabled())
+            {
+                this.WriteEvent(118, clientId);
+            }
+        }
+
+        [Event(119, Level = EventLevel.Informational, Message = "{0}: Unregister MessageHandler done.")]
+        public void UnregisterMessageHandlerStop(string clientId)
+        {
+            if (this.IsEnabled())
+            {
+                this.WriteEvent(119, clientId);
+            }
+        }
     }
 
     internal static class TraceHelper

sdk/servicebus/Microsoft.Azure.ServiceBus/src/QueueClient.cs

@@ -445,6 +445,15 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
             this.InnerReceiver.RegisterMessageHandler(handler, messageHandlerOptions);
         }
 
+        /// <summary>
+        /// Unregister messgae hander from the receiver if there is active message handler registered. 
+        /// </summary>
+        public async Task UnregisterMessageHandler()
+        {
+            this.ThrowIfClosed();
+            await this.InnerReceiver.UnregisterMessageHandler().ConfigureAwait(false);
+        }
+
         /// <summary>
         /// Receive session messages continuously from the queue. Registers a message handler and begins a new thread to receive session-messages.
         /// This handler(<see cref="Func{IMessageSession, Message, CancellationToken, Task}"/>) is awaited on every time a new message is received by the queue client.

sdk/servicebus/Microsoft.Azure.ServiceBus/src/Resources.resx

@@ -192,6 +192,9 @@
   <data name="MessageHandlerAlreadyRegistered" xml:space="preserve">
     <value>A message handler has already been registered.</value>
   </data>
+  <data name="MessageHandlerNotRegisteredYet" xml:space="preserve">
+    <value>A message handler has not been registered yet.</value>
+  </data>
   <data name="SessionHandlerAlreadyRegistered" xml:space="preserve">
     <value>A session handler has already been registered.</value>
   </data>

sdk/servicebus/Microsoft.Azure.ServiceBus/src/SubscriptionClient.cs

@@ -419,6 +419,15 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
             this.InnerSubscriptionClient.InnerReceiver.RegisterMessageHandler(handler, messageHandlerOptions);
         }
 
+        /// <summary>
+        /// Unregister messgae hander from the receiver if there is active message handler registered. 
+        /// </summary>
+        public async Task UnregisterMessageHandler()
+        {
+            this.ThrowIfClosed();
+            await this.InnerSubscriptionClient.InnerReceiver.UnregisterMessageHandler().ConfigureAwait(false);
+        }
+
         /// <summary>
         /// Receive session messages continuously from the queue. Registers a message handler and begins a new thread to receive session-messages.
         /// This handler(<see cref="Func{IMessageSession, Message, CancellationToken, Task}"/>) is awaited on every time a new message is received by the subscription client.
@@ -639,4 +648,4 @@ protected override async Task OnClosingAsync()
             }
         }
     }
-}
+}