From f481c704c930a37bed5096970101d165a23098cd Mon Sep 17 00:00:00 2001
From: Christian Kratky <christian.kratky@googlemail.com>
Date: Tue, 7 Nov 2017 23:30:15 +0100
Subject: [PATCH] Fixed an issue with queued client message in the server.

---
 .../ManagedMqttClientOptionsBuilder.cs        | 49 +++++++++++++++++++
 .../Server/MqttClientPendingMessagesQueue.cs  |  9 ++--
 MQTTnet.Core/TopicFilter.cs                   |  2 +-
 MQTTnet.Core/TopicFilterBuilder.cs            |  2 +-
 4 files changed, 56 insertions(+), 6 deletions(-)
 create mode 100644 MQTTnet.Core/ManagedClient/ManagedMqttClientOptionsBuilder.cs

diff --git a/MQTTnet.Core/ManagedClient/ManagedMqttClientOptionsBuilder.cs b/MQTTnet.Core/ManagedClient/ManagedMqttClientOptionsBuilder.cs
new file mode 100644
index 00000000..3972dcd2
--- /dev/null
+++ b/MQTTnet.Core/ManagedClient/ManagedMqttClientOptionsBuilder.cs
@@ -0,0 +1,49 @@
+using System;
+using MQTTnet.Core.Client;
+
+namespace MQTTnet.Core.ManagedClient
+{
+    public class ManagedMqttClientOptionsBuilder
+    {
+        private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
+
+        public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value)
+        {
+            _options.AutoReconnectDelay = value;
+            return this;
+        }
+
+        public ManagedMqttClientOptionsBuilder WithStorage(IManagedMqttClientStorage value)
+        {
+            _options.Storage = value;
+            return this;
+        }
+
+        public ManagedMqttClientOptionsBuilder WithClientOptions(IMqttClientOptions value)
+        {
+            _options.ClientOptions = value ?? throw new ArgumentNullException(nameof(value));
+            return this;
+        }
+
+        public ManagedMqttClientOptionsBuilder WithClientOptions(Action<MqttClientOptionsBuilder> options)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            var builder = new MqttClientOptionsBuilder();
+            options(builder);
+            _options.ClientOptions = builder.Build();
+
+            return this;
+        }
+
+        public ManagedMqttClientOptions Build()
+        {
+            if (_options.ClientOptions == null)
+            {
+                throw new InvalidOperationException("The ClientOptions cannot be null.");
+            }
+
+            return _options;
+        }
+    }
+}
diff --git a/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs b/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs
index 599f4ee8..6b370db6 100644
--- a/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs
+++ b/MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs
@@ -59,11 +59,12 @@ namespace MQTTnet.Core.Server
 
         private async Task SendPendingPublishPacketAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
         {
-            var packet = _pendingPublishPackets.Take(cancellationToken);
-
+            MqttPublishPacket packet = null;
             try
             {
+                packet = _pendingPublishPackets.Take(cancellationToken);
                 await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
+
                 _logger.LogTrace("Enqueued packet sent (ClientId: {0}).", _session.ClientId);
             }
             catch (Exception exception)
@@ -84,10 +85,10 @@ namespace MQTTnet.Core.Server
                     _logger.LogError(new EventId(), exception, "Sending publish packet failed (ClientId: {0}).", _session.ClientId);
                 }
 
-                if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+                if (packet != null && packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
                 {
                     packet.Dup = true;
-                    _pendingPublishPackets.Add(packet, cancellationToken);
+                    _pendingPublishPackets.Add(packet, CancellationToken.None);
                 }
 
                 _session.Stop();
diff --git a/MQTTnet.Core/TopicFilter.cs b/MQTTnet.Core/TopicFilter.cs
index 3e95afa0..6d5258f4 100644
--- a/MQTTnet.Core/TopicFilter.cs
+++ b/MQTTnet.Core/TopicFilter.cs
@@ -4,7 +4,7 @@ namespace MQTTnet.Core
 {
     public sealed class TopicFilter
     {
-        public TopicFilter(string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public TopicFilter(string topic, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce)
         {
             Topic = topic;
             QualityOfServiceLevel = qualityOfServiceLevel;
diff --git a/MQTTnet.Core/TopicFilterBuilder.cs b/MQTTnet.Core/TopicFilterBuilder.cs
index 90114e7c..bb8416c1 100644
--- a/MQTTnet.Core/TopicFilterBuilder.cs
+++ b/MQTTnet.Core/TopicFilterBuilder.cs
@@ -5,8 +5,8 @@ namespace MQTTnet.Core
 {
     public class TopicFilterBuilder
     {
-        private string _topic;
         private MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
+        private string _topic;
 
         public TopicFilterBuilder WithTopic(string topic)
         {