From 51979fa5a4fd70f5c4ee174284998c1dd6837537 Mon Sep 17 00:00:00 2001
From: Christian <6939810+chkr1011@users.noreply.github.com>
Date: Sun, 19 Oct 2025 19:14:33 +0200
Subject: [PATCH] Call enqueue interceptor for retained messages (#2211)

* Call the event handler also for retained messages

* Update release notes
---
 .../Internal/MqttClientSessionsManager.cs     | 67 +++++++++++--------
 Source/ReleaseNotes.md                        |  1 +
 2 files changed, 40 insertions(+), 28 deletions(-)

diff --git a/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs
index 4b874546..88b46ec4 100644
--- a/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs
+++ b/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs
@@ -189,16 +189,9 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
                         continue;
                     }
 
-                    if (_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
+                    if (await ShouldSkipEnqueue(senderId, session.Id, applicationMessage))
                     {
-                        var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, session.Id, applicationMessage);
-                        await _eventContainer.InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
-
-                        if (!eventArgs.AcceptEnqueue)
-                        {
-                            // Continue checking the other subscriptions
-                            continue;
-                        }
+                        continue;
                     }
 
                     var publishPacketCopy = MqttPublishPacketFactory.Create(applicationMessage);
@@ -311,25 +304,6 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         return Task.FromResult((IList<MqttClientStatus>)result);
     }
 
-    public Task<MqttSessionStatus> GetSessionStatus(string id)
-    {
-        _sessionsManagementLock.EnterReadLock();
-        try
-        {
-            if (!_sessionsStorage.TryGetSession(id, out var session))
-            {
-                throw new InvalidOperationException($"Session with ID '{id}' not found.");
-            }
-
-            var sessionStatus = new MqttSessionStatus(session);
-            return Task.FromResult(sessionStatus);
-        }
-        finally
-        {
-            _sessionsManagementLock.ExitReadLock();
-        }
-    }
-
     public Task<IList<MqttSessionStatus>> GetSessionsStatus()
     {
         var result = new List<MqttSessionStatus>();
@@ -351,6 +325,25 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         return Task.FromResult((IList<MqttSessionStatus>)result);
     }
 
+    public Task<MqttSessionStatus> GetSessionStatus(string id)
+    {
+        _sessionsManagementLock.EnterReadLock();
+        try
+        {
+            if (!_sessionsStorage.TryGetSession(id, out var session))
+            {
+                throw new InvalidOperationException($"Session with ID '{id}' not found.");
+            }
+
+            var sessionStatus = new MqttSessionStatus(session);
+            return Task.FromResult(sessionStatus);
+        }
+        finally
+        {
+            _sessionsManagementLock.ExitReadLock();
+        }
+    }
+
     public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
     {
         MqttConnectedClient connectedClient = null;
@@ -512,6 +505,11 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         {
             foreach (var retainedMessageMatch in subscribeResult.RetainedMessages)
             {
+                if (await ShouldSkipEnqueue(string.Empty, clientId, retainedMessageMatch.ApplicationMessage))
+                {
+                    continue;
+                }
+
                 var publishPacket = MqttPublishPacketFactory.Create(retainedMessageMatch);
                 clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
             }
@@ -736,6 +734,19 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification
         }
     }
 
+    async Task<bool> ShouldSkipEnqueue(string senderId, string clientId, MqttApplicationMessage applicationMessage)
+    {
+        if (!_eventContainer.InterceptingClientEnqueueEvent.HasHandlers)
+        {
+            return false;
+        }
+
+        var eventArgs = new InterceptingClientApplicationMessageEnqueueEventArgs(senderId, clientId, applicationMessage);
+        await _eventContainer.InterceptingClientEnqueueEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+
+        return !eventArgs.AcceptEnqueue;
+    }
+
     async Task<ValidatingConnectionEventArgs> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
     {
         // TODO: Load session items from persisted sessions in the future.
diff --git a/Source/ReleaseNotes.md b/Source/ReleaseNotes.md
index 6481b000..e7cd57d4 100644
--- a/Source/ReleaseNotes.md
+++ b/Source/ReleaseNotes.md
@@ -1,6 +1,7 @@
 * Core: Used new language features across the entire library
 * Core: Performance improvements
 * Server: Improved performance of retained messages when no event handler is attached (#2093, thanks to @zhaowgit)
+* Server: The event `InterceptingClientEnqueue` is now also called for retained messages (BREAKING CHANGE!)
 * Marked all projects as AOT compatible
 * Restored the strong name of the nugets
 * Embedded license file in all nugets (#2197, thanks to @JensNordenbro)