Merge pull request #938 from chkr1011/interceptor-undelivered

Added interceptor for undelivered messages.

Build/MQTTnet.nuspec

@@ -18,6 +18,7 @@
 * [Server] Added state checks (throw if not started etc.) for most server APIs.
 * [Server] Exposed real X509Certificate2 (instead byte array) to TLS options (thanks to @borigas).
 * [Core] Fixed a null reference exception in the MqttTcpChannel with WriteAsync and ReadAsync.
+* [Core] Added server interceptor for undelivered messages (Thanks to @cshark-inator).
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2020</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor</tags> 

Source/MQTTnet/Server/IMqttServerOptions.cs

@@ -25,5 +25,7 @@ namespace MQTTnet.Server
         IMqttServerStorage Storage { get; }
 
         IMqttRetainedMessagesManager RetainedMessagesManager { get; }
+
+        IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttClientSession.cs

@@ -46,17 +46,19 @@ namespace MQTTnet.Server
         /// </summary>
         public IDictionary<object, object> Items { get; }
 
-        public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage)
+        public bool EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage)
         {
             var checkSubscriptionsResult = SubscriptionsManager.CheckSubscriptions(applicationMessage.Topic, applicationMessage.QualityOfServiceLevel);
             if (!checkSubscriptionsResult.IsSubscribed)
             {
-                return;
+                return true;
             }
 
             _logger.Verbose("Queued application message with topic '{0}' (ClientId: {1}).", applicationMessage.Topic, ClientId);
 
             ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
+
+            return false;
         }
 
         public async Task SubscribeAsync(ICollection<MqttTopicFilter> topicFilters)

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -220,12 +220,31 @@ namespace MQTTnet.Server
                     await _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
                 }
 
+                var deliveryCount = 0;
+
                 foreach (var clientSession in _sessions.Values)
                 {
-                    clientSession.EnqueueApplicationMessage(
+                    var isSubscribed = clientSession.EnqueueApplicationMessage(
                         applicationMessage,
                         sender?.ClientId,
                         false);
+
+                    if (isSubscribed)
+                    {
+                        deliveryCount++;
+                    }
+                }
+
+                if (deliveryCount == 0)
+                {
+                    var undeliveredMessageInterceptor = _options.UndeliveredMessageInterceptor;
+
+                    if (undeliveredMessageInterceptor == null)
+                    {
+                        return;
+                    }
+
+                    await undeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(sender?.ClientId, sender?.Session?.Items, applicationMessage));
                 }
             }
             catch (OperationCanceledException)

Source/MQTTnet/Server/MqttServerOptions.cs

@@ -31,5 +31,7 @@ namespace MQTTnet.Server
         public IMqttServerStorage Storage { get; set; }
 
         public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; } = new MqttRetainedMessagesManager();
+
+        public IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; }
     }
 }

Source/MQTTnet/Server/MqttServerOptionsBuilder.cs

@@ -212,5 +212,11 @@ namespace MQTTnet.Server
         {
             return _options;
         }
+
+        public MqttServerOptionsBuilder WithUndeliveredMessageInterceptor(Action<MqttApplicationMessageInterceptorContext> value)
+        {
+            _options.UndeliveredMessageInterceptor = new MqttServerApplicationMessageInterceptorDelegate(value);
+            return this;
+        }
     }
 }

Tests/MQTTnet.Core.Tests/Server_Tests.cs

@@ -1384,5 +1384,33 @@ namespace MQTTnet.Tests
                 Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount);
             }
         }
+
+        [TestMethod]
+        public async Task Intercept_Undelivered()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                var undeliverd = string.Empty;
+                var svr = await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder().WithUndeliveredMessageInterceptor(
+                              context =>
+                              {
+                                  undeliverd = context.ApplicationMessage.Topic;
+                              }));
+
+                var topicAReceived = false;
+                var topicBReceived = false;
+
+                var client = await testEnvironment.ConnectClientAsync();
+
+                await client.SubscribeAsync("b");
+
+                await client.PublishAsync("a", null, MqttQualityOfServiceLevel.ExactlyOnce);
+
+                await Task.Delay(500);
+
+                Assert.AreEqual(undeliverd, "a");
+
+            }
+        }
     }
 }