Expose WithClientMessageQueueInterceptor

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);
+        }
+    }
+}

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;

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);
+            }
+        }
+
     }
 }