Call enqueue interceptor for retained messages (fix) (#2212)

* Call the event handler also for retained messages

* Update release notes

* Add Unit Test

Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs

@@ -189,7 +189,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
                         continue;
                     }
 
-                    if (await ShouldSkipEnqueue(senderId, session.Id, applicationMessage))
+                    if (await _eventContainer.ShouldSkipEnqueue(senderId, session.Id, applicationMessage))
                     {
                         continue;
                     }
@@ -501,18 +501,20 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
 
         var subscribeResult = await clientSession.Subscribe(fakeSubscribePacket, CancellationToken.None).ConfigureAwait(false);
 
-        if (subscribeResult.RetainedMessages != null)
+        if (subscribeResult.RetainedMessages == null)
         {
-            foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
+            return;
-            {
+        }
-                if (await ShouldSkipEnqueue(string.Empty, clientId, retainedMessageMatch.ApplicationMessage))
-                {
-                    continue;
-                }
 
-                var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
+        foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
-                clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
+        {
+            if (await _eventContainer.ShouldSkipEnqueue(string.Empty, clientId, retainedMessageMatch.ApplicationMessage))
+            {
+                continue;
             }
+
+            var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
+            clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
         }
     }
 

Source/MQTTnet.Server/Internal/MqttConnectedClient.cs

@@ -284,13 +284,20 @@ public sealed class MqttConnectedClient : IDisposable
             return;
         }
 
-        if (subscribeResult.RetainedMessages != null)
+        if (subscribeResult.RetainedMessages == null)
         {
-            foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
+            return;
+        }
+
+        foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
+        {
+            if (await _eventContainer.ShouldSkipEnqueue(string.Empty, Id, retainedMessageMatch.ApplicationMessage))
             {
-                var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
+                continue;
-                Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
             }
+
+            var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
+            Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
         }
     }
 

Source/MQTTnet.Server/Internal/MqttServerEventContainer.cs

@@ -6,49 +6,62 @@ using MQTTnet.Internal;
 
 namespace MQTTnet.Server.Internal;
 
-public sealed class MqttServerEventContainer
+public class MqttServerEventContainer
 {
-    public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<ApplicationMessageNotConsumedEventArgs>();
+    public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new();
 
-    public AsyncEvent<ClientAcknowledgedPublishPacketEventArgs> ClientAcknowledgedPublishPacketEvent { get; } = new AsyncEvent<ClientAcknowledgedPublishPacketEventArgs>();
+    public AsyncEvent<ClientAcknowledgedPublishPacketEventArgs> ClientAcknowledgedPublishPacketEvent { get; } = new();
 
-    public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<ClientConnectedEventArgs>();
+    public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new();
 
-    public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<ClientDisconnectedEventArgs>();
+    public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new();
 
-    public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<ClientSubscribedTopicEventArgs>();
+    public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new();
 
-    public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
+    public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new();
 
-    public AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs> InterceptingClientEnqueueEvent { get; } = new AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs>();
+    public AsyncEvent<InterceptingClientApplicationMessageEnqueueEventArgs> InterceptingClientEnqueueEvent { get; } = new();
 
-    public AsyncEvent<ApplicationMessageEnqueuedEventArgs> ApplicationMessageEnqueuedOrDroppedEvent { get; } = new AsyncEvent<ApplicationMessageEnqueuedEventArgs>();
+    public AsyncEvent<ApplicationMessageEnqueuedEventArgs> ApplicationMessageEnqueuedOrDroppedEvent { get; } = new();
 
-    public AsyncEvent<QueueMessageOverwrittenEventArgs> QueuedApplicationMessageOverwrittenEvent { get; } = new AsyncEvent<QueueMessageOverwrittenEventArgs>();
+    public AsyncEvent<QueueMessageOverwrittenEventArgs> QueuedApplicationMessageOverwrittenEvent { get; } = new();
 
-    public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
+    public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new();
 
-    public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
+    public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new();
 
-    public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new AsyncEvent<InterceptingPublishEventArgs>();
+    public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new();
 
-    public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
+    public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new();
 
-    public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
+    public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new();
 
-    public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new AsyncEvent<LoadingRetainedMessagesEventArgs>();
+    public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new();
 
-    public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new AsyncEvent<EventArgs>();
+    public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new();
 
-    public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
+    public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new();
 
-    public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new AsyncEvent<EventArgs>();
+    public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new();
 
-    public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new AsyncEvent<SessionDeletedEventArgs>();
+    public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new();
 
-    public AsyncEvent<EventArgs> StartedEvent { get; } = new AsyncEvent<EventArgs>();
+    public AsyncEvent<EventArgs> StartedEvent { get; } = new();
 
-    public AsyncEvent<EventArgs> StoppedEvent { get; } = new AsyncEvent<EventArgs>();
+    public AsyncEvent<EventArgs> StoppedEvent { get; } = new();
 
-    public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new AsyncEvent<ValidatingConnectionEventArgs>();
+    public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new();
+
+    public async Task<bool> ShouldSkipEnqueue(string senderClientId, string clientId, MqttApplicationMessage applicationMessage)
+    {
+        if (!InterceptingClientEnqueueEvent.HasHandlers)
+        {
+            return false;
+        }
+
+        var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderClientId, clientId, applicationMessage);
+        await InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+
+        return !eventArgs.AcceptEnqueue;
+    }
 }

Source/MQTTnet.Server/Status/MqttSessionStatus.cs

@@ -58,7 +58,7 @@ public sealed class MqttSessionStatus
 
         await packetBusItem.WaitAsync().ConfigureAwait(false);
 
-        var injectResult = new InjectMqttApplicationMessageResult()
+        var injectResult = new InjectMqttApplicationMessageResult
         {
             PacketIdentifier = publishPacket.PacketIdentifier
         };

Source/MQTTnet.Tests/BaseTestClass.cs

@@ -24,4 +24,9 @@ public abstract class BaseTestClass
     {
         return Task.Delay(TimeSpan.FromSeconds(1));
     }
+
+    protected Task MediumTestDelay()
+    {
+        return Task.Delay(TimeSpan.FromSeconds(0.5));
+    }
 }

Source/MQTTnet.Tests/Server/Retained_Messages_Tests.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Tests.Server;
 
 // ReSharper disable InconsistentNaming
 [TestClass]
-public sealed class Retained_Messages_Tests : BaseTestClass
+public class Retained_Messages_Tests : BaseTestClass
 {
     [TestMethod]
     public async Task Clear_Retained_Message_With_Empty_Payload()
@@ -31,9 +31,9 @@ public sealed class Retained_Messages_Tests : BaseTestClass
         var c2 = await testEnvironment.ConnectClient();
         var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
 
-        await Task.Delay(200);
         await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
-        await Task.Delay(500);
+
+        await MediumTestDelay();
 
         messageHandler.AssertReceivedCountEquals(0);
     }
@@ -54,9 +54,9 @@ public sealed class Retained_Messages_Tests : BaseTestClass
         var c2 = await testEnvironment.ConnectClient();
         var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
 
-        await Task.Delay(200);
         await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
-        await Task.Delay(500);
+
+        await MediumTestDelay();
 
         messageHandler.AssertReceivedCountEquals(0);
     }
@@ -82,7 +82,7 @@ public sealed class Retained_Messages_Tests : BaseTestClass
         var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
         await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce });
 
-        await LongTestDelay();
+        await MediumTestDelay();
 
         messageHandler.AssertReceivedCountEquals(1);
 
@@ -110,7 +110,7 @@ public sealed class Retained_Messages_Tests : BaseTestClass
         var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
         await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
 
-        await LongTestDelay();
+        await MediumTestDelay();
 
         messageHandler.AssertReceivedCountEquals(1);
 
@@ -131,7 +131,7 @@ public sealed class Retained_Messages_Tests : BaseTestClass
         var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
         await c2.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("retained_other").Build());
 
-        await Task.Delay(500);
+        await MediumTestDelay();
 
         messageHandler.AssertReceivedCountEquals(0);
     }
@@ -178,10 +178,10 @@ public sealed class Retained_Messages_Tests : BaseTestClass
         var c2 = await testEnvironment.ConnectClient();
         var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
 
-        await Task.Delay(200);
         // Using QoS 2 will lead to 1 instead because the publish was made with QoS level 1 (see 3.8.4 SUBSCRIBE Actions)!
         await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce });
-        await Task.Delay(500);
+
+        await MediumTestDelay();
 
         messageHandler.AssertReceivedCountEquals(1);
     }
@@ -248,4 +248,33 @@ public sealed class Retained_Messages_Tests : BaseTestClass
 
         Assert.IsTrue(connectResult.RetainAvailable);
     }
+
+    [TestMethod]
+    public async Task Skip_Enqueue_Of_Retained_Message()
+    {
+        using var testEnvironment = CreateTestEnvironment();
+
+        var server = await testEnvironment.StartServer();
+        server.InterceptingClientEnqueueAsync += e =>
+        {
+            e.AcceptEnqueue = false;
+
+            return Task.CompletedTask;
+        };
+
+        var c1 = await testEnvironment.ConnectClient();
+
+        await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+
+        await c1.DisconnectAsync();
+
+        var c2 = await testEnvironment.ConnectClient();
+        var messageHandler = testEnvironment.CreateApplicationMessageHandler(c2);
+
+        await c2.SubscribeAsync(new MqttTopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+
+        await MediumTestDelay();
+
+        messageHandler.AssertReceivedCountEquals(0);
+    }
 }