Copy the PublishPacket and then queue it

Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs

@@ -2,6 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System.Buffers;
 using System.Collections;
 using System.Collections.Concurrent;
 using MQTTnet.Adapter;
@@ -155,9 +156,11 @@ public async Task<DispatchApplicationMessageResult> DispatchApplicationMessage(
             var matchingSubscribersCount = 0;
             try
             {
+                var applicationMessageCopy = new Lazy<MqttApplicationMessage>(() => applicationMessage.Clone());
                 if (applicationMessage.Retain)
                 {
-                    await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
+                    // applicationMessage must be copied
+                    await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage, applicationMessageCopy).ConfigureAwait(false);
                 }
 
                 List<MqttSession> subscriberSessions;
@@ -200,7 +203,8 @@ public async Task<DispatchApplicationMessageResult> DispatchApplicationMessage(
                         }
                     }
 
-                    var publishPacketCopy = MqttPublishPacketFactory.Create(applicationMessage);
+                    // applicationMessage must be copied
+                    var publishPacketCopy = MqttPublishPacketFactory.Create(applicationMessageCopy.Value);
                     publishPacketCopy.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
                     publishPacketCopy.SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers;
 
@@ -506,6 +510,7 @@ public Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters)
         return GetClientSession(clientId).Unsubscribe(fakeUnsubscribePacket, CancellationToken.None);
     }
 
+
     MqttConnectedClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
     {
         return new MqttConnectedClient(connectPacket, channelAdapter, session, _options, _eventContainer, this, _rootLogger);
@@ -676,39 +681,39 @@ static bool ShouldPersistSession(MqttConnectedClient connectedClient)
         switch (connectedClient.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion)
         {
             case MqttProtocolVersion.V500:
-            {
-                // MQTT 5.0 section 3.1.2.11.2
-                // The Client and Server MUST store the Session State after the Network Connection is closed if the Session Expiry Interval is greater than 0 [MQTT-3.1.2-23].
-                //
-                // A Client that only wants to process messages while connected will set the Clean Start to 1 and set the Session Expiry Interval to 0.
-                // It will not receive Application Messages published before it connected and has to subscribe afresh to any topics that it is interested
-                // in each time it connects.
-
-                var effectiveSessionExpiryInterval = connectedClient.DisconnectPacket?.SessionExpiryInterval ?? 0U;
-                if (effectiveSessionExpiryInterval == 0U)
                 {
-                    // From RFC: If the Session Expiry Interval is absent, the Session Expiry Interval in the CONNECT packet is used.
-                    effectiveSessionExpiryInterval = connectedClient.ConnectPacket.SessionExpiryInterval;
-                }
+                    // MQTT 5.0 section 3.1.2.11.2
+                    // The Client and Server MUST store the Session State after the Network Connection is closed if the Session Expiry Interval is greater than 0 [MQTT-3.1.2-23].
+                    //
+                    // A Client that only wants to process messages while connected will set the Clean Start to 1 and set the Session Expiry Interval to 0.
+                    // It will not receive Application Messages published before it connected and has to subscribe afresh to any topics that it is interested
+                    // in each time it connects.
+
+                    var effectiveSessionExpiryInterval = connectedClient.DisconnectPacket?.SessionExpiryInterval ?? 0U;
+                    if (effectiveSessionExpiryInterval == 0U)
+                    {
+                        // From RFC: If the Session Expiry Interval is absent, the Session Expiry Interval in the CONNECT packet is used.
+                        effectiveSessionExpiryInterval = connectedClient.ConnectPacket.SessionExpiryInterval;
+                    }
 
-                return effectiveSessionExpiryInterval != 0U;
-            }
+                    return effectiveSessionExpiryInterval != 0U;
+                }
 
             case MqttProtocolVersion.V311:
-            {
-                // MQTT 3.1.1 section 3.1.2.4: persist only if 'not CleanSession'
-                //
-                // If CleanSession is set to 1, the Client and Server MUST discard any previous Session and start a new one.
-                // This Session lasts as long as the Network Connection. State data associated with this Session MUST NOT be
-                // reused in any subsequent Session [MQTT-3.1.2-6].
+                {
+                    // MQTT 3.1.1 section 3.1.2.4: persist only if 'not CleanSession'
+                    //
+                    // If CleanSession is set to 1, the Client and Server MUST discard any previous Session and start a new one.
+                    // This Session lasts as long as the Network Connection. State data associated with this Session MUST NOT be
+                    // reused in any subsequent Session [MQTT-3.1.2-6].
 
-                return !connectedClient.ConnectPacket.CleanSession;
-            }
+                    return !connectedClient.ConnectPacket.CleanSession;
+                }
 
             case MqttProtocolVersion.V310:
-            {
-                return true;
-            }
+                {
+                    return true;
+                }
 
             default:
                 throw new NotSupportedException();

Source/MQTTnet.Server/Internal/MqttRetainedMessagesManager.cs

@@ -50,7 +50,7 @@ public async Task Start()
             }
         }
 
-        public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage)
+        public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage, Lazy<MqttApplicationMessage> applicationMessageCopy)
         {
             ArgumentNullException.ThrowIfNull(applicationMessage);
 
@@ -61,10 +61,7 @@ public async Task UpdateMessage(string clientId, MqttApplicationMessage applicat
 
                 lock (_messages)
                 {
-                    var payload = applicationMessage.Payload;
-                    var hasPayload = payload.Length > 0;
-
-                    if (!hasPayload)
+                    if (applicationMessage.Payload.IsEmpty)
                     {
                         saveIsRequired = _messages.Remove(applicationMessage.Topic);
                         _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
@@ -73,15 +70,15 @@ public async Task UpdateMessage(string clientId, MqttApplicationMessage applicat
                     {
                         if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage))
                         {
-                            _messages[applicationMessage.Topic] = applicationMessage;
+                            _messages[applicationMessage.Topic] = applicationMessageCopy.Value;
                             saveIsRequired = true;
                         }
                         else
                         {
                             if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel ||
-                                !MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload))
+                                !MqttMemoryHelper.SequenceEqual(existingMessage.Payload, applicationMessage.Payload))
                             {
-                                _messages[applicationMessage.Topic] = applicationMessage;
+                                _messages[applicationMessage.Topic] = applicationMessageCopy.Value;
                                 saveIsRequired = true;
                             }
                         }
@@ -99,7 +96,7 @@ public async Task UpdateMessage(string clientId, MqttApplicationMessage applicat
                 {
                     using (await _storageAccessLock.EnterAsync().ConfigureAwait(false))
                     {
-                        var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave);
+                        var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessageCopy.Value, messagesForSave);
                         await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                     }
                 }

Source/MQTTnet.Server/MqttServer.cs

@@ -352,7 +352,7 @@ public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage)
         ThrowIfDisposed();
         ThrowIfNotStarted();
 
-        return _retainedMessagesManager?.UpdateMessage(string.Empty, retainedMessage);
+        return _retainedMessagesManager?.UpdateMessage(string.Empty, retainedMessage, new Lazy<MqttApplicationMessage>(retainedMessage));
     }
 
     protected override void Dispose(bool disposing)

Source/MQTTnet.Tests/Mockups/MqttApplicationMessageReceived.cs

@@ -0,0 +1,4 @@
+namespace MQTTnet.Tests.Mockups
+{
+    public record MqttApplicationMessageReceived(string ClientId, MqttApplicationMessage ApplicationMessage);
+}

Source/MQTTnet.Tests/Mockups/TestApplicationMessageReceivedHandler.cs

@@ -14,13 +14,16 @@ namespace MQTTnet.Tests.Mockups
 {
     public sealed class TestApplicationMessageReceivedHandler
     {
-        readonly List<MqttApplicationMessageReceivedEventArgs> _receivedEventArgs = new List<MqttApplicationMessageReceivedEventArgs>();
+        readonly IMqttClient _mqttClient;
+        readonly List<MqttApplicationMessageReceived> _receivedEventArgs = new();
+
 
         public TestApplicationMessageReceivedHandler(IMqttClient mqttClient)
         {
             ArgumentNullException.ThrowIfNull(mqttClient);
 
             mqttClient.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;
+            _mqttClient = mqttClient;
         }
 
         public int Count
@@ -34,7 +37,7 @@ public int Count
             }
         }
 
-        public List<MqttApplicationMessageReceivedEventArgs> ReceivedEventArgs
+        public List<MqttApplicationMessageReceived> ReceivedEventArgs
         {
             get
             {
@@ -72,7 +75,11 @@ Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs e
         {
             lock (_receivedEventArgs)
             {
-                _receivedEventArgs.Add(eventArgs);
+                var applicationMessage = _mqttClient.Options.ReceivedApplicationMessageQueueable
+                    ? eventArgs.ApplicationMessage
+                    : eventArgs.ApplicationMessage.Clone();
+
+                _receivedEventArgs.Add(new MqttApplicationMessageReceived(eventArgs.ClientId, applicationMessage));
             }
 
             return CompletedTask.Instance;

Source/MQTTnet.Tests/Server/QoS_Tests.cs

@@ -113,13 +113,24 @@ public async Task Fire_Event_On_Client_Acknowledges_QoS_2()
 
     [TestMethod]
     public async Task Preserve_Message_Order_For_Queued_Messages()
+    {
+        await Preserve_Message_Order_For_Queued_Messages(receivedPublishPacketQueueable: true);
+    }
+
+    [TestMethod]
+    public async Task Preserve_Message_Order_For_Queued_Messages_NoQueue()
+    {
+        await Preserve_Message_Order_For_Queued_Messages(receivedPublishPacketQueueable: false);
+    }
+
+    private async Task Preserve_Message_Order_For_Queued_Messages(bool receivedPublishPacketQueueable)
     {
         using (var testEnvironment = CreateTestEnvironment())
         {
             var server = await testEnvironment.StartServer(o => o.WithPersistentSessions());
 
             // Create a session which will contain the messages.
-            var dummyClient = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false));
+            var dummyClient = await testEnvironment.ConnectClient(o => o.WithClientId("A").WithCleanSession(false).WithReceivedApplicationMessageQueueable(receivedPublishPacketQueueable));
             await dummyClient.SubscribeAsync("#", MqttQualityOfServiceLevel.AtLeastOnce);
             dummyClient.Dispose();
 

Source/MQTTnet/MqttApplicationMessage.cs

@@ -8,6 +8,7 @@
 using System;
 using System.Buffers;
 using System.Collections.Generic;
+using System.Linq;
 
 namespace MQTTnet
 {
@@ -142,5 +143,29 @@ public ArraySegment<byte> PayloadSegment
         ///     Hint: MQTT 5 feature only.
         /// </summary>
         public List<MqttUserProperty> UserProperties { get; set; }
+
+        /// <summary>
+        /// Deep clone all fields.
+        /// </summary>
+        /// <returns></returns>
+        public MqttApplicationMessage Clone()
+        {
+            return new MqttApplicationMessage
+            {
+                ContentType = ContentType,
+                CorrelationData = CorrelationData == default ? default : CorrelationData.ToArray(),
+                Dup = Dup,
+                MessageExpiryInterval = MessageExpiryInterval,
+                Payload = Payload.IsEmpty ? default : new ReadOnlySequence<byte>(Payload.ToArray()),
+                PayloadFormatIndicator = PayloadFormatIndicator,
+                QualityOfServiceLevel = QualityOfServiceLevel,
+                Retain = Retain,
+                ResponseTopic = ResponseTopic,
+                Topic = Topic,
+                UserProperties = UserProperties?.Select(u => u.Clone()).ToList(),
+                SubscriptionIdentifiers = SubscriptionIdentifiers?.ToList(),
+                TopicAlias = TopicAlias
+            };
+        }
     }
 }