Add functionality to unregister message handler to servicebus sdk #6360

rangp · 2019-05-21T16:00:38Z

The MessageReceiver class and thus the topic and queue clients have the method RegisterMessageHandler that allow registration of a message handler that will be used to process incoming messages without the need to manually receive or complete those messages.

However there is no method to unregister the message handler and the only way to stop automatic message processing seems to be calling CloseAsync which will stop message reception but will also immediately close the service bus connection, making it impossible to complete messages that are currently processed, if the processing is long running (And PeekLock is used). Cancelling the processing upon CloseAsync is not an option in my use-case since the processing is kind of a transaction scope and has to be completed fully or not at all.

Hence I feel the need for functionality that allows one to cancel automatic message reception while still keeping the service bus connection open. This pull request introduces the method UnregisterMessageHandler implemented in a way that will solve my problem and I think it would be a consistent thing to add to the API, that could also be used to change message handlers without closing the service bus connection.

Kind regards
Peter

msftclas · 2019-05-21T16:00:58Z

All CLA requirements met.

azuresdkci · 2019-05-21T16:02:02Z

Can one of the admins verify this patch?

[skip ci]

nemakam · 2019-07-12T01:22:01Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+                        await Task.Delay(TimeSpan.FromSeconds(5));
+                    }
+
+                    Assert.True(count < messageCount);


messageCount [](start = 40, length = 12)

Should this be count <= maxConcurrentCalls?

nemakam · 2019-07-12T01:49:16Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+                            messageReceiver.UnregisterMessageHandler();
+
+                            // recursively register a new handler and expect it to handle the open messages 
+                            HandlerRegistration(false);


HandlerRegistration(false); [](start = 28, length = 27)

Am I missing something? How is this even working? For maxConcurrentCall of say 5, there are 5 different instances of this handler code which are trying to call RegisterMessageHandler. And there is always a race condition where two of them try to register without unregistering in between. And that would throw.

nemakam · 2019-07-12T01:50:20Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+
+                        Interlocked.Increment(ref count);
+                    },
+                    new MessageHandlerOptions(ExceptionReceivedHandler) { MaxConcurrentCalls = maxConcurrentCalls, AutoComplete = autoComplete });


ExceptionReceivedHandler [](start = 46, length = 24)

I am thinking for these tests we need to make sure there are no exceptions. We need to have another variable and Assert that there are no exceptions received during this time.

nemakam · 2019-07-12T01:50:37Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+                            messageReceiver.UnregisterMessageHandler();
+
+                            // Simulate "long running" task
+                            Thread.Sleep(1000);


Thread.Sleep(1000); [](start = 28, length = 19)

await Task.Delay()

nemakam · 2019-07-12T01:51:47Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+
+                    // Wait for the OnMessage Tasks to finish
+                    var stopwatch = Stopwatch.StartNew();
+                    while (stopwatch.Elapsed.TotalSeconds <= 30)


30 [](start = 61, length = 2)

4 different permutations of this test, which essentially means 2 mins added run time. Not a blocker, but maybe consider optimizing this to a smaller number.

nemakam

🕐

nemakam · 2019-07-12T01:54:42Z

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

@@ -445,6 +445,16 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
            this.InnerReceiver.RegisterMessageHandler(handler, messageHandlerOptions);
        }

+        /// <summary>
+        /// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.


continuuous [](start = 24, length = 11)

typo

nemakam · 2019-07-12T01:56:54Z

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

@@ -445,6 +445,16 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
            this.InnerReceiver.RegisterMessageHandler(handler, messageHandlerOptions);
        }

+        /// <summary>
+        /// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.
+        /// <remarks>Register a message handler first, using <see cref="RegisterMessageHandler(Func{Message, CancellationToken, Task}, Func{ExceptionReceivedEventArgs, Task})"/> 


remarks [](start = 13, length = 7)

Also add a remark mentioning the behavior when this is called multiple times

The same in other clients

In reply to: 302802451 [](ancestors = 302802451)

Also, looks like there is a difference in the xml doc between different files. Could you make sure all xmldocs of this method have the same comment

In reply to: 302802465 [](ancestors = 302802465,302802451)

nemakam · 2019-07-12T01:57:58Z

sealed class MessageReceivePump

This needs to be done for SessionReceivePump as well to keep it consistent.

Refers to: sdk/servicebus/Microsoft.Azure.ServiceBus/src/MessageReceivePump.cs:13 in 8637512. [](commit_id = 8637512, deletion_comment = False)

nemakam · 2019-07-12T01:59:14Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+
+            // Wait for the OnMessage Tasks to finish
+            var stopwatch = Stopwatch.StartNew();
+            while (stopwatch.Elapsed.TotalSeconds <= 60)


60 [](start = 53, length = 2)

Again, could you reduce this further (and enable prefetch on the receiver so that messages are received faster)

nemakam · 2019-07-12T01:59:42Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+                }
+                await Task.Delay(TimeSpan.FromSeconds(5));
+            }
+            Assert.True(count == messageCount);


True [](start = 19, length = 4)

Assert.Equal

nemakam · 2019-07-12T02:00:07Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+            bool autoComplete,
+            int messageCount)
+            {
+            var count = 0;


nit: could you please take care of indentation?

nemakam · 2019-07-12T02:00:38Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/SenderReceiverClientTestBase.cs

+
+                        if (messageReceiver.ReceiveMode == ReceiveMode.PeekLock && !autoComplete)
+                        {
+                                                    // Completion should still work, even after message handler is stopped


// Completion should still work, even after message handler is stopped [](start = 52, length = 70)

nit: indentation

nemakam · 2019-07-12T02:03:49Z

sdk/servicebus/Microsoft.Azure.ServiceBus/tests/MessageReceiverTests.cs

+    using Microsoft.Azure.ServiceBus.Core;
+    using Xunit;
+
+    public sealed class MessageReceiverTests : SenderReceiverClientTestBase


MessageReceiverTests [](start = 24, length = 20)

We don't have a test which ensures that once Unregister is called, there are no more messages that are received, i.e, ensure no handlers are called after that

nemakam · 2019-07-12T02:06:29Z

@rangp, thanks for the PR. The overall design looks good. There are couple of small changes that I have requested.

SeanFeldman · 2019-08-07T22:15:12Z

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

@@ -419,6 +419,16 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
            this.InnerSubscriptionClient.InnerReceiver.RegisterMessageHandler(handler, messageHandlerOptions);
        }

+        /// <summary>
+        /// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.


typo

Suggested change

/// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

/// Cancels the continuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

SeanFeldman · 2019-08-07T22:15:29Z

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

@@ -63,6 +63,14 @@ public interface IReceiverClient : IClientEntity
        /// <remarks>Enable prefetch to speed up the receive rate.</remarks>
        void RegisterMessageHandler(Func<Message, CancellationToken, Task> handler, MessageHandlerOptions messageHandlerOptions);

+        /// <summary>
+        /// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.


typo

Suggested change

/// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

/// Cancels the continuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

SeanFeldman · 2019-08-07T22:18:11Z

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

@@ -899,6 +903,27 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
            this.OnMessageHandler(messageHandlerOptions, handler);
        }

+        /// <summary>
+        /// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.


typo

Suggested change

/// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

/// Cancels the continuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

SeanFeldman · 2019-08-07T22:23:32Z

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

@@ -445,6 +445,16 @@ public void RegisterMessageHandler(Func<Message, CancellationToken, Task> handle
            this.InnerReceiver.RegisterMessageHandler(handler, messageHandlerOptions);
        }

+        /// <summary>
+        /// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.


typo

Suggested change

/// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

/// Cancels the continuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

SeanFeldman · 2019-08-07T22:42:37Z

@nemakam given that IReceiverClient is changing, this will be a breaking change.
Should this be indicated somehow?

nemakam · 2019-08-08T21:19:45Z

@SeanFeldman , yeah. Maybe we could initially put this into the MessageReceiver class and not the interface and have a separate PR for changes in interface which can be checked in whenever the next major version is to be released.

@rangp , could you please separate out the PR with changes in the IMessageReceiver in a separate PR so that this change need not wait for the next major release.

pakrym · 2019-11-04T23:47:50Z

This PR is stale and will be closed, please reopen if there are new updates.

DorothySun216 · 2021-02-24T20:32:02Z

We published this functionality in 5.0.0: Enable a way to Unregister Message Handler and Session Handler PR 14021

rangp added 2 commits May 21, 2019 17:31

added functionality to unregister message handlers

60ca43a

Merge branch 'master' of https://github.com/rangp/azure-sdk-for-net

8f8ff42

dsgouda requested a review from AlexGhiondea May 21, 2019 16:59

dsgouda added the needs-review label May 21, 2019

rangp added 2 commits June 5, 2019 10:24

Set up CI with Azure Pipelines

18e0a41

[skip ci]

Delete azure-pipelines.yml

8637512

ramya-rao-a requested review from nemakam and binzywu and removed request for AlexGhiondea June 28, 2019 01:23

ramya-rao-a added the Service Bus label Jun 28, 2019

nemakam reviewed Jul 12, 2019

View reviewed changes

nemakam suggested changes Jul 12, 2019

View reviewed changes

nemakam reviewed Jul 12, 2019

View reviewed changes

SeanFeldman reviewed Aug 7, 2019

View reviewed changes

pakrym closed this Nov 4, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add functionality to unregister message handler to servicebus sdk #6360

Add functionality to unregister message handler to servicebus sdk #6360

rangp commented May 21, 2019

msftclas commented May 21, 2019 •

edited

Loading

azuresdkci commented May 21, 2019

nemakam Jul 12, 2019 •

edited

Loading

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam left a comment

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam commented Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam Jul 12, 2019

nemakam commented Jul 12, 2019

SeanFeldman Aug 7, 2019

SeanFeldman Aug 7, 2019

SeanFeldman Aug 7, 2019

SeanFeldman Aug 7, 2019

SeanFeldman commented Aug 7, 2019

nemakam commented Aug 8, 2019

pakrym commented Nov 4, 2019

DorothySun216 commented Feb 24, 2021

	/// Cancels the continuuous reception of messages without closing the underlying service bus connection and unregisters the message handler.
	/// Cancels the continuous reception of messages without closing the underlying service bus connection and unregisters the message handler.

Add functionality to unregister message handler to servicebus sdk #6360

Add functionality to unregister message handler to servicebus sdk #6360

Conversation

rangp commented May 21, 2019

msftclas commented May 21, 2019 • edited Loading

azuresdkci commented May 21, 2019

nemakam Jul 12, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nemakam left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nemakam commented Jul 12, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nemakam commented Jul 12, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SeanFeldman commented Aug 7, 2019

nemakam commented Aug 8, 2019

pakrym commented Nov 4, 2019

DorothySun216 commented Feb 24, 2021

msftclas commented May 21, 2019 •

edited

Loading

nemakam Jul 12, 2019 •

edited

Loading