Merge pull request #979 from msallin/bugfix/978/ClientMessageQueueInterceptor

Bugfix/978/client message queue interceptor

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -392,22 +392,18 @@ namespace MQTTnet.Server
                     }
 
                     publishPacket = _dataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage);
-                    publishPacket.QualityOfServiceLevel = queuedApplicationMessage.QualityOfServiceLevel;
+                    publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel;
 
                     // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9].
                     publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage;
 
-                    if (publishPacket.QualityOfServiceLevel > 0)
-                    {
-                        publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-                    }
-
                     if (_serverOptions.ClientMessageQueueInterceptor != null)
                     {
                         var context = new MqttClientMessageQueueInterceptorContext(
                             queuedApplicationMessage.SenderClientId,
                             ClientId,
-                            queuedApplicationMessage.ApplicationMessage);
+                            queuedApplicationMessage.ApplicationMessage,
+                            queuedApplicationMessage.SubscriptionQualityOfServiceLevel);
 
                         if (_serverOptions.ClientMessageQueueInterceptor != null)
                         {
@@ -421,7 +417,12 @@ namespace MQTTnet.Server
 
                         publishPacket.Topic = context.ApplicationMessage.Topic;
                         publishPacket.Payload = context.ApplicationMessage.Payload;
-                        publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel;
+                        publishPacket.QualityOfServiceLevel = context.SubscriptionQualityOfServiceLevel;
+                    }
+
+                    if (publishPacket.QualityOfServiceLevel > 0)
+                    {
+                        publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
                     }
 
                     if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)

Source/MQTTnet/Server/MqttClientMessageQueueInterceptorContext.cs

@@ -1,12 +1,15 @@
-namespace MQTTnet.Server
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server
 {
     public class MqttClientMessageQueueInterceptorContext
     {
-        public MqttClientMessageQueueInterceptorContext(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage)
+        public MqttClientMessageQueueInterceptorContext(string senderClientId, string receiverClientId, MqttApplicationMessage applicationMessage, MqttQualityOfServiceLevel subscriptionQualityOfServiceLevel)
         {
             SenderClientId = senderClientId;
             ReceiverClientId = receiverClientId;
             ApplicationMessage = applicationMessage;
+            SubscriptionQualityOfServiceLevel = subscriptionQualityOfServiceLevel;
         }
 
         public string SenderClientId { get; }
@@ -16,5 +19,7 @@
         public MqttApplicationMessage ApplicationMessage { get; set; }
 
         public bool AcceptEnqueue { get; set; } = true;
+
+        public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; }
     }
 }

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/MqttClientSessionApplicationMessagesQueue.cs

@@ -27,7 +27,7 @@ namespace MQTTnet.Server
             {
                 ApplicationMessage = applicationMessage,
                 SenderClientId = senderClientId,
-                QualityOfServiceLevel = qualityOfServiceLevel,
+                SubscriptionQualityOfServiceLevel = qualityOfServiceLevel,
                 IsRetainedMessage = isRetainedMessage
             });
         }

Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs

@@ -1,4 +1,5 @@
-using MQTTnet.Protocol;
+using System;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
@@ -10,7 +11,14 @@ namespace MQTTnet.Server
 
         public bool IsRetainedMessage { get; set; }
 
-        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
+        public MqttQualityOfServiceLevel SubscriptionQualityOfServiceLevel { get; set; }
+
+        [Obsolete("Use 'SubscriptionQualityOfServiceLevel' instead.")]
+        public MqttQualityOfServiceLevel QualityOfServiceLevel
+        {
+            get => SubscriptionQualityOfServiceLevel;
+            set => SubscriptionQualityOfServiceLevel = value;
+        }
 
         public bool IsDuplicate { get; set; }
     }

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,68 @@ 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);
+            }
+        }
+
+        [TestMethod]
+        public async Task Intercept_ClientMessageQueue_Different_QoS_Of_Subscription_And_Message()
+        {
+            const string topic = "a";
+
+            using (var testEnvironment = new TestEnvironment(TestContext))
+            {
+                await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder()
+                    .WithClientMessageQueueInterceptor(c => { })); // Interceptor does nothing but has to be present.
+
+                bool receivedMessage = false;
+                var client = await testEnvironment.ConnectClientAsync();
+                client.UseApplicationMessageReceivedHandler(c =>
+                {
+                    receivedMessage = true;
+                });
+
+                await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce);
+
+                await client.PublishAsync(new MqttApplicationMessage{ Topic = topic, QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+
+                await Task.Delay(500);
+
+                Assert.IsTrue(receivedMessage);
+            }
+        }
     }
 }