implement intercept publish message event on ManagedMqttClient (#1915)

* implement intercept publish message event on ManagedMqttClient

* rename ShouldPublish to ACceptPublish per PR feedback

Source/MQTTnet.Extensions.ManagedClient/InterceptingPublishMessageEventArgs.cs

@@ -0,0 +1,20 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+
+namespace MQTTnet.Extensions.ManagedClient
+{
+    public sealed class InterceptingPublishMessageEventArgs : EventArgs
+    {
+        public InterceptingPublishMessageEventArgs(ManagedMqttApplicationMessage applicationMessage)
+        {
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+        }
+
+        public ManagedMqttApplicationMessage ApplicationMessage { get; }
+
+        public bool AcceptPublish { get; set; } = true;
+    }
+}

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -19,6 +19,7 @@ namespace MQTTnet.Extensions.ManagedClient
 {
     public sealed class ManagedMqttClient : Disposable, IManagedMqttClient
     {
+        readonly AsyncEvent<InterceptingPublishMessageEventArgs> _interceptingPublishMessageEvent = new AsyncEvent<InterceptingPublishMessageEventArgs>();
         readonly AsyncEvent<ApplicationMessageProcessedEventArgs> _applicationMessageProcessedEvent = new AsyncEvent<ApplicationMessageProcessedEventArgs>();
         readonly AsyncEvent<ApplicationMessageSkippedEventArgs> _applicationMessageSkippedEvent = new AsyncEvent<ApplicationMessageSkippedEventArgs>();
         readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
@@ -88,6 +89,14 @@ namespace MQTTnet.Extensions.ManagedClient
             remove => _applicationMessageProcessedEvent.RemoveHandler(value);
         }
 
+
+        public event Func<InterceptingPublishMessageEventArgs, Task> InterceptPublishMessageAsync
+        {
+            add => _interceptingPublishMessageEvent.AddHandler(value);
+            remove => _interceptingPublishMessageEvent.RemoveHandler(value);
+        }
+
+
         public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
         {
             add => InternalClient.ApplicationMessageReceivedAsync += value;
@@ -700,9 +709,20 @@ namespace MQTTnet.Extensions.ManagedClient
         async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
         {
             Exception transmitException = null;
+            bool acceptPublish = true;
             try
             {
-                await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
+                if (_interceptingPublishMessageEvent.HasHandlers)
+                {
+                    var interceptEventArgs = new InterceptingPublishMessageEventArgs(message);
+                    await _interceptingPublishMessageEvent.InvokeAsync(interceptEventArgs).ConfigureAwait(false);
+                    acceptPublish = interceptEventArgs.AcceptPublish;
+                }
+
+                if (acceptPublish)
+                {
+                    await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
+                }
 
                 using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync
                 {

Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs

@@ -437,6 +437,109 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient
             }
         }
 
+
+        [TestMethod]
+        public async Task ManagedClients_CanInterceptPublishedMessage_PreventingPublish()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var receivedMessagesCount = 0;
+                var interceptedMessagesCount = 0;
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .Build();
+
+                var sendingClient = testEnvironment.CreateClient();
+                var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
+
+                await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
+
+                // Wait until the managed client is fully set up and running.
+                await Task.Delay(1000);
+
+                var receivingClient = await testEnvironment.ConnectClient();
+
+                receivingClient.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return CompletedTask.Instance;
+                };
+                sendingManagedClient.InterceptPublishMessageAsync += e =>
+                {
+                    Interlocked.Increment(ref interceptedMessagesCount);
+                    e.AcceptPublish = false;
+                    return CompletedTask.Instance;
+                };
+
+                await receivingClient.SubscribeAsync("My/last/will");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                Assert.AreEqual(0, receivedMessagesCount);
+                Assert.AreEqual(1, interceptedMessagesCount);
+            }
+        }
+
+        [TestMethod]
+        public async Task ManagedClients_CanInterceptPublishedMessage_AllowingPublish()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var receivedMessagesCount = 0;
+                var interceptedMessagesCount = 0;
+
+                var clientOptions = new MqttClientOptionsBuilder()
+                    .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                    .Build();
+
+                var sendingClient = testEnvironment.CreateClient();
+                var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger);
+
+                await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build());
+
+                // Wait until the managed client is fully set up and running.
+                await Task.Delay(1000);
+
+                var receivingClient = await testEnvironment.ConnectClient();
+
+                receivingClient.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return CompletedTask.Instance;
+                };
+                sendingManagedClient.InterceptPublishMessageAsync += e =>
+                {
+                    Interlocked.Increment(ref interceptedMessagesCount);
+                    e.AcceptPublish = true;
+                    return CompletedTask.Instance;
+                };
+
+                await receivingClient.SubscribeAsync("My/last/will");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                await sendingManagedClient.EnqueueAsync("My/last/will", "hello world");
+
+                // Wait for arrival of the will message at the receiver.
+                await Task.Delay(5000);
+
+                Assert.AreEqual(1, receivedMessagesCount);
+                Assert.AreEqual(1, interceptedMessagesCount);
+            }
+        }
+
         // This case also serves as a regression test for the previous behavior which re-published
         // each and every existing subscriptions with every new subscription that was made
         // (causing performance problems and having the visible symptom of retained messages being received again)