From ccc275902fdf98f3e8b36846b56c30b819fe00d2 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Wed, 28 Nov 2018 21:35:50 +0100 Subject: [PATCH] Fix wrong retain flag when distributing messages. --- Build/MQTTnet.AspNetCore.nuspec | 2 +- Build/MQTTnet.Extensions.ManagedClient.nuspec | 2 +- Build/MQTTnet.Extensions.Rpc.nuspec | 2 +- Build/MQTTnet.nuspec | 5 +- Source/MQTTnet/Server/MqttClientSession.cs | 9 +++- .../Server/MqttClientSessionsManager.cs | 7 ++- Tests/MQTTnet.Core.Tests/MqttClientTests.cs | 51 +++++++++++++++++-- Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 14 +++-- 8 files changed, 77 insertions(+), 15 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index d68b4414..d59a5246 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -14,7 +14,7 @@ Copyright Christian Kratky 2016-2018 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 - + diff --git a/Build/MQTTnet.Extensions.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index 91b6a237..6fb244b0 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -14,7 +14,7 @@ Copyright Christian Kratky 2016-2018 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 - + diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index 36f8965c..b77569ef 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -14,7 +14,7 @@ Copyright Christian Kratky 2016-2018 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 - + diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 83e04775..f120bb85 100644 --- a/Build/MQTTnet.nuspec +++ b/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 concurrent sends with Aspnetcore.Connections.Abstractions based transport. +* [Server] Fixed cpu spike in case a client disconnects (issue 421). +* [Server] Fixed concurrent sends with Aspnetcore.Connections.Abstractions based transport. +* [Server] Fixed wrong retain flag when distributing application messages (thanks to @trev0115). Copyright Christian Kratky 2016-2018 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 diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 131cd397..12045692 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/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); } } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 8bc800ea..28acc79d 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/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) diff --git a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs index 212e1abb..7d68bef1 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/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(); + + 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()); diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 2a48e993..b86c7b5a 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/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(); 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]