Fix wrong retain flag when distributing messages.

Build/MQTTnet.AspNetCore.nuspec

@@ -14,7 +14,7 @@
     <copyright>Copyright Christian Kratky 2016-2018</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</tags>
     <dependencies>
-      <dependency id="MQTTnet" version="2.8.5-rc4" />
+      <dependency id="MQTTnet" version="2.8.5-rc5" />
       <dependency id="Microsoft.AspNetCore.Connections.Abstractions" version="2.1.3" />
       <dependency id="Microsoft.AspNetCore.WebSockets" version="2.1.1" />
       <dependency id="Microsoft.Extensions.Hosting.Abstractions" version="2.1.1" />

Build/MQTTnet.Extensions.ManagedClient.nuspec

@@ -14,7 +14,7 @@
     <copyright>Copyright Christian Kratky 2016-2018</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</tags> 
     <dependencies>
-      <dependency id="MQTTnet" version="2.8.5-rc4" />
+      <dependency id="MQTTnet" version="2.8.5-rc5" />
     </dependencies>
   </metadata>
 

Build/MQTTnet.Extensions.Rpc.nuspec

@@ -14,7 +14,7 @@
     <copyright>Copyright Christian Kratky 2016-2018</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</tags> 
     <dependencies>
-      <dependency id="MQTTnet" version="2.8.5-rc4" />
+      <dependency id="MQTTnet" version="2.8.5-rc5" />
     </dependencies>
   </metadata>
 

Build/MQTTnet.nuspec

@@ -17,8 +17,9 @@
 * [ManagedClient] Fixed an issue which deletes the wrong message from the internal queue (thanks to @PaulFake).
 * [Server] Added new method which exposes all retained messages.
 * [Server] Removed (wrong) setter from the server options interface.
-* [Server] fixed cpu spike in case a client disconnects (issue 421).
+* [Server] Fixed cpu spike in case a client disconnects (issue 421).
-* [Server] fixed concurrent sends with Aspnetcore.Connections.Abstractions based transport.
+* [Server] Fixed concurrent sends with Aspnetcore.Connections.Abstractions based transport.
+* [Server] Fixed wrong retain flag when distributing application messages (thanks to @trev0115).
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2018</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</tags> 

Source/MQTTnet/Server/MqttClientSession.cs

@@ -227,7 +227,7 @@ namespace MQTTnet.Server
                 Topic = publishPacket.Topic,
                 Payload = publishPacket.Payload,
                 QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel,
-                Retain = false,
+                Retain = publishPacket.Retain,
                 Dup = false
             };
 
@@ -371,7 +371,12 @@ namespace MQTTnet.Server
             var retainedMessages = _retainedMessagesManager.GetSubscribedMessages(topicFilters);
             foreach (var applicationMessage in retainedMessages)
             {
-                EnqueueApplicationMessage(null, applicationMessage.ToPublishPacket());
+                var publishPacket = applicationMessage.ToPublishPacket();
+
+                // Set the retain flag to true according to [MQTT-3.3.1-8].
+                publishPacket.Retain = true;
+
+                EnqueueApplicationMessage(null, publishPacket);
             }
         }
 

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -185,7 +185,12 @@ namespace MQTTnet.Server
 
                 foreach (var clientSession in GetSessions())
                 {
-                    clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket());
+                    var publishPacket = applicationMessage.ToPublishPacket();
+
+                    // Set the retain flag to true according to [MQTT-3.3.1-9].
+                    publishPacket.Retain = false;
+
+                    clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, publishPacket);
                 }
             }
             catch (OperationCanceledException)

Tests/MQTTnet.Core.Tests/MqttClientTests.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Net.Sockets;
 using System.Threading;
 using System.Threading.Tasks;
@@ -8,6 +10,7 @@ using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Packets;
+using MQTTnet.Server;
 
 namespace MQTTnet.Core.Tests
 {
@@ -39,6 +42,45 @@ namespace MQTTnet.Core.Tests
             Assert.IsInstanceOfType(ex.InnerException, typeof(SocketException));
         }
 
+        [TestMethod]
+        public async Task ClientPublish()
+        {
+            var server = new MqttFactory().CreateMqttServer();
+            
+            try
+            {
+                var receivedMessages = new List<MqttApplicationMessage>();
+
+                await server.StartAsync(new MqttServerOptions());
+
+                var client1 = new MqttFactory().CreateMqttClient();
+                client1.ApplicationMessageReceived += (_, e) =>
+                {
+                    lock (receivedMessages)
+                    {
+                        receivedMessages.Add(e.ApplicationMessage);
+                    }
+                };
+
+                await client1.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").Build());
+                await client1.SubscribeAsync("a");
+
+                var client2 = new MqttFactory().CreateMqttClient();
+                await client2.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").Build());
+                var message = new MqttApplicationMessageBuilder().WithTopic("a").WithRetainFlag().Build();
+                await client2.PublishAsync(message);
+
+                await Task.Delay(500);
+
+                Assert.AreEqual(1, receivedMessages.Count);
+                Assert.IsFalse(receivedMessages.First().Retain); // Must be false even if set above!
+            }
+            finally
+            {
+                await server.StopAsync();
+            }
+        }
+
 #if DEBUG
         [TestMethod]
         public async Task ClientCleanupOnAuthentificationFails()
@@ -50,11 +92,12 @@ namespace MQTTnet.Core.Tests
 
             Task.Run(async () => {
                 var connect = await channel2.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None);
-                await channel2.SendPacketAsync(new MqttConnAckPacket() { ConnectReturnCode = Protocol.MqttConnectReturnCode.ConnectionRefusedNotAuthorized }, CancellationToken.None);
+                await channel2.SendPacketAsync(new MqttConnAckPacket
+                {
+                    ConnectReturnCode = Protocol.MqttConnectReturnCode.ConnectionRefusedNotAuthorized
+                }, CancellationToken.None);
             });
-
+            
-
-
             var fake = new TestMqttCommunicationAdapterFactory(channel);
 
             var client = new MqttClient(fake, new MqttNetLogger());

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -430,7 +430,7 @@ namespace MQTTnet.Core.Tests
             var serverAdapter = new TestMqttServerAdapter();
             var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
 
-            var receivedMessagesCount = 0;
+            var receivedMessages = new List<MqttApplicationMessage>();
             try
             {
                 await s.StartAsync(new MqttServerOptions());
@@ -440,7 +440,14 @@ namespace MQTTnet.Core.Tests
                 await c1.DisconnectAsync();
 
                 var c2 = await serverAdapter.ConnectTestClient("c2");
-                c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
+                c2.ApplicationMessageReceived += (_, e) =>
+                {
+                    lock (receivedMessages)
+                    {
+                        receivedMessages.Add(e.ApplicationMessage);
+                    }
+                };
+
                 await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("retained").Build());
 
                 await Task.Delay(500);
@@ -450,7 +457,8 @@ namespace MQTTnet.Core.Tests
                 await s.StopAsync();
             }
 
-            Assert.AreEqual(1, receivedMessagesCount);
+            Assert.AreEqual(1, receivedMessages.Count);
+            Assert.IsTrue(receivedMessages.First().Retain);
         }
 
         [TestMethod]