Merge branch 'develop'

dotnet · Aug 5, 2017 · 0172606 · 0172606
2 parents 36abf84 + 65c2a0f
commit 0172606
Show file tree

Hide file tree

Showing 34 changed files with 356 additions and 246 deletions.
diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec
@@ -10,7 +10,11 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
-    <releaseNotes>* 
+    <releaseNotes>* [MqttServer] Added support for publishing application messages
+* [Core] Fixed QoS level 2 handling
+* [Core] Performance optimizations
+* [MqttClient/MqttServer] Errors while handline application messages are now catched and traced
+* [MqttClient/MqttServer] Added interfaces
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2017</copyright>
     <tags>MQTT MQTTClient MQTTServer MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Queue Hardware</tags> 

diff --git a/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs b/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs
@@ -8,7 +8,7 @@ namespace MQTTnet
 {
     public class MqttClientFactory
     {
-        public MqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

diff --git a/Frameworks/MQTTnet.NetFramework/MqttServerFactory.cs b/Frameworks/MQTTnet.NetFramework/MqttServerFactory.cs
@@ -8,7 +8,7 @@ namespace MQTTnet
 {
     public class MqttServerFactory
     {
-        public MqttServer CreateMqttServer(MqttServerOptions options)
+        public IMqttServer CreateMqttServer(MqttServerOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

diff --git a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs
@@ -8,7 +8,7 @@ namespace MQTTnet
 {
     public class MqttClientFactory
     {
-        public MqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

diff --git a/Frameworks/MQTTnet.NetStandard/MqttServerFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttServerFactory.cs
@@ -8,7 +8,7 @@ namespace MQTTnet
 {
     public class MqttServerFactory
     {
-        public MqttServer CreateMqttServer(MqttServerOptions options)
+        public IMqttServer CreateMqttServer(MqttServerOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs
@@ -8,7 +8,7 @@ namespace MQTTnet
 {
     public class MqttClientFactory
     {
-        public MqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttServerFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttServerFactory.cs
@@ -8,7 +8,7 @@ namespace MQTTnet
 {
     public class MqttServerFactory
     {
-        public MqttServer CreateMqttServer(MqttServerOptions options)
+        public IMqttServer CreateMqttServer(MqttServerOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

diff --git a/MQTTnet.Core/Client/IMqttClient.cs b/MQTTnet.Core/Client/IMqttClient.cs
@@ -0,0 +1,24 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using MQTTnet.Core.Packets;
+
+namespace MQTTnet.Core.Client
+{
+    public interface IMqttClient
+    {
+        bool IsConnected { get; }
+
+        event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+        event EventHandler Connected;
+        event EventHandler Disconnected;
+
+        Task ConnectAsync(MqttApplicationMessage willApplicationMessage = null);
+        Task DisconnectAsync();
+        Task PublishAsync(MqttApplicationMessage applicationMessage);
+        Task<IList<MqttSubscribeResult>> SubscribeAsync(IList<TopicFilter> topicFilters);
+        Task<IList<MqttSubscribeResult>> SubscribeAsync(params TopicFilter[] topicFilters);
+        Task Unsubscribe(IList<string> topicFilters);
+        Task Unsubscribe(params string[] topicFilters);
+    }
+}
diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs
@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -13,10 +12,9 @@
 
 namespace MQTTnet.Core.Client
 {
-    public class MqttClient
+    public class MqttClient : IMqttClient
     {
-        private readonly ConcurrentDictionary<ushort, MqttPublishPacket> _pendingExactlyOncePublishPackets = new ConcurrentDictionary<ushort, MqttPublishPacket>();
-        private readonly HashSet<ushort> _processedPublishPackets = new HashSet<ushort>();
+        private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
 
         private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
         private readonly MqttClientOptions _options;
@@ -63,7 +61,6 @@ public async Task ConnectAsync(MqttApplicationMessage willApplicationMessage = n
 
             _cancellationTokenSource = new CancellationTokenSource();
             _latestPacketIdentifier = 0;
-            _processedPublishPackets.Clear();
             _packetDispatcher.Reset();
             IsConnected = true;
 
@@ -105,6 +102,7 @@ public async Task<IList<MqttSubscribeResult>> SubscribeAsync(IList<TopicFilter>
         {
             if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
             if (!topicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
+
             ThrowIfNotConnected();
 
             var subscribePacket = new MqttSubscribePacket
@@ -154,6 +152,7 @@ public async Task PublishAsync(MqttApplicationMessage applicationMessage)
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
+                // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
                 await SendAsync(publishPacket);
             }
             else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
@@ -164,8 +163,8 @@ public async Task PublishAsync(MqttApplicationMessage applicationMessage)
             else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
             {
                 publishPacket.PacketIdentifier = GetNewPacketIdentifier();
-                await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket);
-                await SendAsync(publishPacket.CreateResponse<MqttPubCompPacket>());
+                var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket);
+                await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>());
             }
         }
 
@@ -208,14 +207,12 @@ private Task ProcessReceivedPacketAsync(MqttBasePacket mqttPacket)
                     return DisconnectAsync();
                 }
 
-                var publishPacket = mqttPacket as MqttPublishPacket;
-                if (publishPacket != null)
+                if (mqttPacket is MqttPublishPacket publishPacket)
                 {
                     return ProcessReceivedPublishPacket(publishPacket);
                 }
 
-                var pubRelPacket = mqttPacket as MqttPubRelPacket;
-                if (pubRelPacket != null)
+                if (mqttPacket is MqttPubRelPacket pubRelPacket)
                 {
                     return ProcessReceivedPubRelPacket(pubRelPacket);
                 }
@@ -232,13 +229,16 @@ private Task ProcessReceivedPacketAsync(MqttBasePacket mqttPacket)
 
         private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
         {
-            if (publishPacket.QualityOfServiceLevel != MqttQualityOfServiceLevel.AtMostOnce)
+            var applicationMessage = publishPacket.ToApplicationMessage();
+
+            try
             {
-                _processedPublishPackets.Add(publishPacket.PacketIdentifier);
+                ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while handling application message.");    
             }
-
-            var applicationMessage = publishPacket.ToApplicationMessage();
-            ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
         }
 
         private Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
@@ -257,7 +257,13 @@ private Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
             {
-                _pendingExactlyOncePublishPackets[publishPacket.PacketIdentifier] = publishPacket;
+                // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
+                lock (_unacknowledgedPublishPackets)
+                {
+                    _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
+                }
+
+                FireApplicationMessageReceivedEvent(publishPacket);
                 return SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
             }
 
@@ -266,15 +272,12 @@ private Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
 
         private async Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket)
         {
-            MqttPublishPacket originalPublishPacket;
-            if (!_pendingExactlyOncePublishPackets.TryRemove(pubRelPacket.PacketIdentifier, out originalPublishPacket))
+            lock (_unacknowledgedPublishPackets)
             {
-                throw new MqttCommunicationException();
+                _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
             }
-
-            await SendAsync(originalPublishPacket.CreateResponse<MqttPubCompPacket>());
-
-            FireApplicationMessageReceivedEvent(originalPublishPacket);
+
+            await SendAsync(pubRelPacket.CreateResponse<MqttPubCompPacket>());
         }
 
         private Task SendAsync(MqttBasePacket packet)
@@ -292,8 +295,8 @@ bool ResponsePacketSelector(MqttBasePacket p)
                     return false;
                 }
 
-                var pi1 = requestPacket as IPacketWithIdentifier;
-                var pi2 = p as IPacketWithIdentifier;
+                var pi1 = requestPacket as IMqttPacketWithIdentifier;
+                var pi2 = p as IMqttPacketWithIdentifier;
 
                 if (pi1 != null && pi2 != null)
                 {

diff --git a/...re/Packets/IPacketWithPacketIdentifier.cs → ...Core/Packets/IMqttPacketWithIdentifier.cs b/...re/Packets/IPacketWithPacketIdentifier.cs → ...Core/Packets/IMqttPacketWithIdentifier.cs
@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Packets
 {
-    public interface IPacketWithIdentifier
+    public interface IMqttPacketWithIdentifier
     {
         ushort PacketIdentifier { get; set; }
     }

diff --git a/MQTTnet.Core/Packets/MqttBasePacket.cs b/MQTTnet.Core/Packets/MqttBasePacket.cs
@@ -1,25 +1,6 @@
-using System;
-
-namespace MQTTnet.Core.Packets
+namespace MQTTnet.Core.Packets
 {
     public abstract class MqttBasePacket
     {
-        public TResponsePacket CreateResponse<TResponsePacket>()
-        {
-            var responsePacket = Activator.CreateInstance<TResponsePacket>();
-            var responsePacketWithIdentifier = responsePacket as IPacketWithIdentifier;
-            if (responsePacketWithIdentifier != null)
-            {
-                var requestPacketWithIdentifier = this as IPacketWithIdentifier;
-                if (requestPacketWithIdentifier == null)
-                {
-                    throw new InvalidOperationException("Response packet has PacketIdentifier but request packet does not.");
-                }
-
-                responsePacketWithIdentifier.PacketIdentifier = requestPacketWithIdentifier.PacketIdentifier;
-            }
-
-            return responsePacket;
-        }
     }
 }
diff --git a/MQTTnet.Core/Packets/MqttBasePublishPacket.cs b/MQTTnet.Core/Packets/MqttBasePublishPacket.cs
@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Packets
 {
-    public class MqttBasePublishPacket : MqttBasePacket, IPacketWithIdentifier
+    public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
     }

diff --git a/MQTTnet.Core/Packets/MqttPacketExtensions.cs b/MQTTnet.Core/Packets/MqttPacketExtensions.cs
@@ -0,0 +1,27 @@
+using System;
+
+namespace MQTTnet.Core.Packets
+{
+    public static class MqttPacketExtensions
+    {
+        public static TResponsePacket CreateResponse<TResponsePacket>(this MqttBasePacket packet)
+        {
+            if (packet == null) throw new ArgumentNullException(nameof(packet));
+
+            var responsePacket = Activator.CreateInstance<TResponsePacket>();
+
+            if (responsePacket is IMqttPacketWithIdentifier responsePacketWithIdentifier)
+            {
+                var requestPacketWithIdentifier = packet as IMqttPacketWithIdentifier;
+                if (requestPacketWithIdentifier == null)
+                {
+                    throw new InvalidOperationException("Response packet has PacketIdentifier but request packet does not.");
+                }
+
+                responsePacketWithIdentifier.PacketIdentifier = requestPacketWithIdentifier.PacketIdentifier;
+            }
+
+            return responsePacket;
+        }
+    }
+}
diff --git a/MQTTnet.Core/Packets/MqttSubAckPacket.cs b/MQTTnet.Core/Packets/MqttSubAckPacket.cs
@@ -4,7 +4,7 @@
 
 namespace MQTTnet.Core.Packets
 {
-    public sealed class MqttSubAckPacket : MqttBasePacket, IPacketWithIdentifier
+    public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
 

diff --git a/MQTTnet.Core/Packets/MqttSubscribePacket.cs b/MQTTnet.Core/Packets/MqttSubscribePacket.cs
@@ -3,7 +3,7 @@
 
 namespace MQTTnet.Core.Packets
 {
-    public sealed class MqttSubscribePacket : MqttBasePacket, IPacketWithIdentifier
+    public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
 

diff --git a/MQTTnet.Core/Packets/MqttUnsubAckPacket.cs b/MQTTnet.Core/Packets/MqttUnsubAckPacket.cs
@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Packets
 {
-    public sealed class MqttUnsubAckPacket : MqttBasePacket, IPacketWithIdentifier
+    public sealed class MqttUnsubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
     }

diff --git a/MQTTnet.Core/Packets/MqttUnsubscribe.cs b/MQTTnet.Core/Packets/MqttUnsubscribe.cs
@@ -2,7 +2,7 @@
 
 namespace MQTTnet.Core.Packets
 {
-    public sealed class MqttUnsubscribePacket : MqttBasePacket, IPacketWithIdentifier
+    public sealed class MqttUnsubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
 

diff --git a/MQTTnet.Core/Serializer/ByteReader.cs b/MQTTnet.Core/Serializer/ByteReader.cs
@@ -24,7 +24,7 @@ public bool Read()
             return result;
         }
 
-        public byte Read(int count)
+        public int Read(int count)
         {
             if (_index + count > 8)
             {
@@ -42,7 +42,7 @@ public byte Read(int count)
                 _index++;
             }
 
-            return (byte)result;
+            return result;
         }
     }
 }