From f1f9924bce3932488c46c3a9ff4fe6012bdc7519 Mon Sep 17 00:00:00 2001
From: Marc Sallin <marc.sallin@outlook.com>
Date: Tue, 25 Aug 2020 00:48:19 +0200
Subject: [PATCH] Expose WithClientMessageQueueInterceptor

---
 ...ttClientMessageQueueInterceptorDelegate.cs | 31 ++++++++++++++++
 .../Server/MqttServerOptionsBuilder.cs        | 12 ++++++
 Tests/MQTTnet.Core.Tests/Server_Tests.cs      | 37 +++++++++++++++++++
 3 files changed, 80 insertions(+)
 create mode 100644 Source/MQTTnet/Server/MqttClientMessageQueueInterceptorDelegate.cs

diff --git a/Source/MQTTnet/Server/MqttClientMessageQueueInterceptorDelegate.cs b/Source/MQTTnet/Server/MqttClientMessageQueueInterceptorDelegate.cs
new file mode 100644
index 00000000..2cb7fc33
--- /dev/null
+++ b/Source/MQTTnet/Server/MqttClientMessageQueueInterceptorDelegate.cs
@@ -0,0 +1,31 @@
+using System;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Server
+{
+    public class MqttClientMessageQueueInterceptorDelegate : IMqttServerClientMessageQueueInterceptor
+    {
+        private readonly Func<MqttClientMessageQueueInterceptorContext, Task> _callback;
+
+        public MqttClientMessageQueueInterceptorDelegate(Action<MqttClientMessageQueueInterceptorContext> callback)
+        {
+            if (callback == null) throw new ArgumentNullException(nameof(callback));
+
+            _callback = context =>
+            {
+                callback(context);
+                return Task.FromResult(0);
+            };
+        }
+
+        public MqttClientMessageQueueInterceptorDelegate(Func<MqttClientMessageQueueInterceptorContext, Task> callback)
+        {
+            _callback = callback ?? throw new ArgumentNullException(nameof(callback));
+        }
+
+        public Task InterceptClientMessageQueueEnqueueAsync(MqttClientMessageQueueInterceptorContext context)
+        {
+            return _callback(context);
+        }
+    }
+}
\ No newline at end of file
diff --git a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs
index 19ce1453..38c91830 100644
--- a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs
+++ b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs
@@ -175,6 +175,18 @@ namespace MQTTnet.Server
             return this;
         }
 
+        public MqttServerOptionsBuilder WithClientMessageQueueInterceptor(IMqttServerClientMessageQueueInterceptor value)
+        {
+            _options.ClientMessageQueueInterceptor = value;
+            return this;
+        }
+
+        public MqttServerOptionsBuilder WithClientMessageQueueInterceptor(Action<MqttClientMessageQueueInterceptorContext> value)
+        {
+            _options.ClientMessageQueueInterceptor = new MqttClientMessageQueueInterceptorDelegate(value);
+            return this;
+        }
+
         public MqttServerOptionsBuilder WithSubscriptionInterceptor(IMqttServerSubscriptionInterceptor value)
         {
             _options.SubscriptionInterceptor = value;
diff --git a/Tests/MQTTnet.Core.Tests/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_Tests.cs
index 1ff7bfab..9ea8627e 100644
--- a/Tests/MQTTnet.Core.Tests/Server_Tests.cs
+++ b/Tests/MQTTnet.Core.Tests/Server_Tests.cs
@@ -1415,5 +1415,42 @@ namespace MQTTnet.Tests
                 Assert.AreEqual("a", undeliverd);
             }
         }
+
+        [TestMethod]
+        public async Task Intercept_ClientMessageQueue()
+        {
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder()
+                    .WithClientMessageQueueInterceptor(c => c.ApplicationMessage.Topic = "a"));
+
+                var topicAReceived = false;
+                var topicBReceived = false;
+
+                var client = await testEnvironment.ConnectClientAsync();
+                client.UseApplicationMessageReceivedHandler(c =>
+                {
+                    if (c.ApplicationMessage.Topic == "a")
+                    {
+                        topicAReceived = true;
+                    }
+                    else if (c.ApplicationMessage.Topic == "b")
+                    {
+                        topicBReceived = true;
+                    }
+                });
+
+                await client.SubscribeAsync("a");
+                await client.SubscribeAsync("b");
+
+                await client.PublishAsync("b");
+
+                await Task.Delay(500);
+
+                Assert.IsTrue(topicAReceived);
+                Assert.IsFalse(topicBReceived);
+            }
+        }
+
     }
 }