Fix wrong QoS level handling for server.

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.1.4</version>
+    <version>2.1.5.1</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,12 +10,10 @@
     <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>* [.NET Standard] Added support for .NET Standard 1.3 (Thanks to 1iveowl)
-* [Server] Added an even which allows processing of all received messages from the clients
-* [Server] Fixed an issue when stopping the server
+    <releaseNotes>* [Server] Fixed wrong handling of QoS levels
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2017</copyright>
-    <tags>MQTT MQTTClient MQTTServer MQTTBroker Broker</tags> 
+    <tags>MQTT MQTTClient MQTTServer MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Queue Hardware</tags> 
     <dependencies>
 
       <group targetFramework="netstandard1.3">

Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs

@@ -86,7 +86,7 @@ namespace MQTTnet.Implementations
                 try
                 {
                     var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null);
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception) when (!(exception is ObjectDisposedException))
@@ -107,7 +107,7 @@ namespace MQTTnet.Implementations
                     var sslStream = new SslStream(new NetworkStream(clientSocket));
                     await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false);
                     
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception)

Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs

@@ -8,11 +8,11 @@ namespace MQTTnet
 {
     public class MqttClientFactory
     {
-        public MqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new DefaultMqttV311PacketSerializer()));
+            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new MqttV311PacketSerializer()));
         }
     }
 }

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));
 

Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs

@@ -11,5 +11,5 @@ using System.Runtime.InteropServices;
 [assembly: AssemblyCulture("")]
 [assembly: ComVisible(false)]
 [assembly: Guid("a480ef90-0eaa-4d9a-b271-47a9c47f6f7d")]
-[assembly: AssemblyVersion("2.1.4.0")]
-[assembly: AssemblyFileVersion("2.1.4.0")]
+[assembly: AssemblyVersion("2.1.5.1")]
+[assembly: AssemblyFileVersion("2.1.5.1")]

Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs

@@ -84,7 +84,7 @@ namespace MQTTnet.Implementations
                 try
                 {
                     var clientSocket = await _defaultEndpointSocket.AcceptAsync();
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception)
@@ -105,7 +105,7 @@ namespace MQTTnet.Implementations
                     var sslStream = new SslStream(new NetworkStream(clientSocket));
                     await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false);
                     
-                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new DefaultMqttV311PacketSerializer());
+                    var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttV311PacketSerializer());
                     ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                 }
                 catch (Exception exception)

Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj

@@ -4,8 +4,8 @@
     <TargetFramework>netstandard1.3</TargetFramework>
     <AssemblyName>MQTTnet</AssemblyName>
     <RootNamespace>MQTTnet</RootNamespace>
-    <AssemblyVersion>2.1.4.0</AssemblyVersion>
-    <FileVersion>2.1.4.0</FileVersion>
+    <AssemblyVersion>2.1.5.1</AssemblyVersion>
+    <FileVersion>2.1.5.1</FileVersion>
     <Version>0.0.0.0</Version>
     <Company />
     <Product />

Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs

@@ -8,11 +8,11 @@ namespace MQTTnet
 {
     public class MqttClientFactory
     {
-        public MqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new DefaultMqttV311PacketSerializer()));
+            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new MqttV311PacketSerializer()));
         }
     }
 }

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));
 

Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs

@@ -52,7 +52,7 @@ namespace MQTTnet.Implementations
         {
             try
             {
-                var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(args.Socket), new DefaultMqttV311PacketSerializer());
+                var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(args.Socket), new MqttV311PacketSerializer());
                 ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(args.Socket.Information.RemoteAddress.ToString(), clientAdapter));
             }
             catch (Exception exception)

Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs

@@ -8,11 +8,11 @@ namespace MQTTnet
 {
     public class MqttClientFactory
     {
-        public MqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient(MqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new DefaultMqttV311PacketSerializer()));
+            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new MqttV311PacketSerializer()));
         }
     }
 }

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));
 

Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs

@@ -10,5 +10,5 @@ using System.Runtime.InteropServices;
 [assembly: AssemblyTrademark("")]
 [assembly: AssemblyCulture("")]
 [assembly: ComVisible(false)]
-[assembly: AssemblyVersion("2.1.4.0")]
-[assembly: AssemblyFileVersion("2.1.4.0")]
+[assembly: AssemblyVersion("2.1.5.1")]
+[assembly: AssemblyFileVersion("2.1.5.1")]

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);
+    }
+}

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 @@ using MQTTnet.Core.Protocol;
 
 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 @@ namespace MQTTnet.Core.Client
 
             _cancellationTokenSource = new CancellationTokenSource();
             _latestPacketIdentifier = 0;
-            _processedPublishPackets.Clear();
             _packetDispatcher.Reset();
             IsConnected = true;
 
@@ -105,6 +102,7 @@ namespace MQTTnet.Core.Client
         {
             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 @@ namespace MQTTnet.Core.Client
 
             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 @@ namespace MQTTnet.Core.Client
             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 @@ namespace MQTTnet.Core.Client
                     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 @@ namespace MQTTnet.Core.Client
 
         private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
         {
-            if (publishPacket.QualityOfServiceLevel != MqttQualityOfServiceLevel.AtMostOnce)
-            {
-                _processedPublishPackets.Add(publishPacket.PacketIdentifier);
-            }
-
             var applicationMessage = publishPacket.ToApplicationMessage();
-            ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
+
+            try
+            {
+                ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while handling application message.");    
+            }
         }
 
         private Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
@@ -257,24 +257,27 @@ namespace MQTTnet.Core.Client
 
             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 });
             }
 
-            throw new InvalidOperationException();
+            throw new MqttCommunicationException("Received a not supported QoS level.");
         }
 
         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 @@ namespace MQTTnet.Core.Client
                     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)
                 {

MQTTnet.Core/MQTTnet.Core.csproj

@@ -16,8 +16,8 @@
     <PackageIconUrl></PackageIconUrl>
     <RepositoryUrl></RepositoryUrl>
     <PackageTags></PackageTags>
-    <FileVersion>2.1.4.0</FileVersion>
-    <AssemblyVersion>2.1.4.0</AssemblyVersion>
+    <FileVersion>2.1.5.1</FileVersion>
+    <AssemblyVersion>2.1.5.1</AssemblyVersion>
     <PackageLicenseUrl></PackageLicenseUrl>
   </PropertyGroup>
 

MQTTnet.Core/Packets/IMqttPacketWithIdentifier.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Packets
 {
-    public interface IPacketWithIdentifier
+    public interface IMqttPacketWithIdentifier
     {
         ushort PacketIdentifier { get; set; }
     }

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;
-        }
     }
 }

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; }
     }

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;
+        }
+    }
+}

MQTTnet.Core/Packets/MqttSubAckPacket.cs

@@ -4,7 +4,7 @@ using MQTTnet.Core.Protocol;
 
 namespace MQTTnet.Core.Packets
 {
-    public sealed class MqttSubAckPacket : MqttBasePacket, IPacketWithIdentifier
+    public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
 

MQTTnet.Core/Packets/MqttSubscribePacket.cs

@@ -3,7 +3,7 @@ using System.Linq;
 
 namespace MQTTnet.Core.Packets
 {
-    public sealed class MqttSubscribePacket : MqttBasePacket, IPacketWithIdentifier
+    public sealed class MqttSubscribePacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
         

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; }
     }

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; }
         

MQTTnet.Core/Serializer/ByteReader.cs

@@ -24,7 +24,7 @@ namespace MQTTnet.Core.Serializer
             return result;
         }
 
-        public byte Read(int count)
+        public int Read(int count)
         {
             if (_index + count > 8)
             {
@@ -42,7 +42,7 @@ namespace MQTTnet.Core.Serializer
                 _index++;
             }
 
-            return (byte)result;
+            return result;
         }
     }
 }

MQTTnet.Core/Serializer/MqttPacketWriter.cs

@@ -11,39 +11,6 @@ namespace MQTTnet.Core.Serializer
     {
         private readonly MemoryStream _buffer = new MemoryStream(512);
 
-        public void InjectFixedHeader(byte fixedHeader)
-        {
-            if (_buffer.Length == 0)
-            {
-                Write(fixedHeader);
-                Write(0);
-                return;
-            }
-
-            var backupBuffer = _buffer.ToArray();
-            var remainingLength = (int)_buffer.Length;
-
-            _buffer.SetLength(0);
-
-            _buffer.WriteByte(fixedHeader);
-
-            // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
-            var x = remainingLength;
-            do
-            {
-                var encodedByte = x % 128;
-                x = x / 128;
-                if (x > 0)
-                {
-                    encodedByte = encodedByte | 128;
-                }
-
-                _buffer.WriteByte((byte)encodedByte);
-            } while (x > 0);
-
-            _buffer.Write(backupBuffer, 0, backupBuffer.Length);
-        }
-
         public void InjectFixedHeader(MqttControlPacketType packetType, byte flags = 0)
         {
             var fixedHeader = (byte)((byte)packetType << 4);
@@ -101,5 +68,38 @@ namespace MQTTnet.Core.Serializer
         {
             _buffer?.Dispose();
         }
+
+        private void InjectFixedHeader(byte fixedHeader)
+        {
+            if (_buffer.Length == 0)
+            {
+                Write(fixedHeader);
+                Write(0);
+                return;
+            }
+
+            var backupBuffer = _buffer.ToArray();
+            var remainingLength = (int)_buffer.Length;
+
+            _buffer.SetLength(0);
+
+            _buffer.WriteByte(fixedHeader);
+
+            // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
+            var x = remainingLength;
+            do
+            {
+                var encodedByte = x % 128;
+                x = x / 128;
+                if (x > 0)
+                {
+                    encodedByte = encodedByte | 128;
+                }
+
+                _buffer.WriteByte((byte)encodedByte);
+            } while (x > 0);
+
+            _buffer.Write(backupBuffer, 0, backupBuffer.Length);
+        }
     }
 }

MQTTnet.Core/Serializer/MqttV311PacketSerializer.cs

@@ -9,93 +9,81 @@ using MQTTnet.Core.Protocol;
 
 namespace MQTTnet.Core.Serializer
 {
-    public sealed class DefaultMqttV311PacketSerializer : IMqttPacketSerializer
+    public sealed class MqttV311PacketSerializer : IMqttPacketSerializer
     {
+        private static readonly byte[] MqttV311Prefix = Encoding.UTF8.GetBytes("MQTT");
+
         public Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
             if (destination == null) throw new ArgumentNullException(nameof(destination));
 
-            var connectPacket = packet as MqttConnectPacket;
-            if (connectPacket != null)
+            if (packet is MqttConnectPacket connectPacket)
             {
                 return SerializeAsync(connectPacket, destination);
             }
 
-            var connAckPacket = packet as MqttConnAckPacket;
-            if (connAckPacket != null)
+            if (packet is MqttConnAckPacket connAckPacket)
             {
                 return SerializeAsync(connAckPacket, destination);
             }
 
-            var disconnectPacket = packet as MqttDisconnectPacket;
-            if (disconnectPacket != null)
+            if (packet is MqttDisconnectPacket disconnectPacket)
             {
                 return SerializeAsync(disconnectPacket, destination);
             }
 
-            var pingReqPacket = packet as MqttPingReqPacket;
-            if (pingReqPacket != null)
+            if (packet is MqttPingReqPacket pingReqPacket)
             {
                 return SerializeAsync(pingReqPacket, destination);
             }
 
-            var pingRespPacket = packet as MqttPingRespPacket;
-            if (pingRespPacket != null)
+            if (packet is MqttPingRespPacket pingRespPacket)
             {
                 return SerializeAsync(pingRespPacket, destination);
             }
 
-            var publishPacket = packet as MqttPublishPacket;
-            if (publishPacket != null)
+            if (packet is MqttPublishPacket publishPacket)
             {
                 return SerializeAsync(publishPacket, destination);
             }
 
-            var pubAckPacket = packet as MqttPubAckPacket;
-            if (pubAckPacket != null)
+            if (packet is MqttPubAckPacket pubAckPacket)
             {
                 return SerializeAsync(pubAckPacket, destination);
             }
 
-            var pubRecPacket = packet as MqttPubRecPacket;
-            if (pubRecPacket != null)
+            if (packet is MqttPubRecPacket pubRecPacket)
             {
                 return SerializeAsync(pubRecPacket, destination);
             }
 
-            var pubRelPacket = packet as MqttPubRelPacket;
-            if (pubRelPacket != null)
+            if (packet is MqttPubRelPacket pubRelPacket)
             {
                 return SerializeAsync(pubRelPacket, destination);
             }
 
-            var pubCompPacket = packet as MqttPubCompPacket;
-            if (pubCompPacket != null)
+            if (packet is MqttPubCompPacket pubCompPacket)
             {
                 return SerializeAsync(pubCompPacket, destination);
             }
 
-            var subscribePacket = packet as MqttSubscribePacket;
-            if (subscribePacket != null)
+            if (packet is MqttSubscribePacket subscribePacket)
             {
                 return SerializeAsync(subscribePacket, destination);
             }
 
-            var subAckPacket = packet as MqttSubAckPacket;
-            if (subAckPacket != null)
+            if (packet is MqttSubAckPacket subAckPacket)
             {
                 return SerializeAsync(subAckPacket, destination);
             }
 
-            var unsubscribePacket = packet as MqttUnsubscribePacket;
-            if (unsubscribePacket != null)
+            if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
                 return SerializeAsync(unsubscribePacket, destination);
             }
 
-            var unsubAckPacket = packet as MqttUnsubAckPacket;
-            if (unsubAckPacket != null)
+            if (packet is MqttUnsubAckPacket unsubAckPacket)
             {
                 return SerializeAsync(unsubAckPacket, destination);
             }
@@ -206,7 +194,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private async Task<MqttBasePacket> DeserializeUnsubscribeAsync(MqttPacketReader reader)
+        private static async Task<MqttBasePacket> DeserializeUnsubscribeAsync(MqttPacketReader reader)
         {
             var packet = new MqttUnsubscribePacket
             {
@@ -221,7 +209,7 @@ namespace MQTTnet.Core.Serializer
             return packet;
         }
 
-        private async Task<MqttBasePacket> DeserializeSubscribeAsync(MqttPacketReader reader)
+        private static async Task<MqttBasePacket> DeserializeSubscribeAsync(MqttPacketReader reader)
         {
             var packet = new MqttSubscribePacket
             {
@@ -238,7 +226,7 @@ namespace MQTTnet.Core.Serializer
             return packet;
         }
 
-        private async Task<MqttBasePacket> DeserializePublishAsync(MqttPacketReader reader)
+        private static async Task<MqttBasePacket> DeserializePublishAsync(MqttPacketReader reader)
         {
             var fixedHeader = new ByteReader(reader.FixedHeader);
             var retain = fixedHeader.Read();
@@ -266,13 +254,10 @@ namespace MQTTnet.Core.Serializer
             return packet;
         }
 
-        private async Task<MqttBasePacket> DeserializeConnectAsync(MqttPacketReader reader)
+        private static async Task<MqttBasePacket> DeserializeConnectAsync(MqttPacketReader reader)
         {
-            var packet = new MqttConnectPacket();
-
-            await reader.ReadRemainingDataByteAsync();
-            await reader.ReadRemainingDataByteAsync();
-
+            await reader.ReadRemainingDataAsync(2); // Skip 2 bytes
+            
             var protocolName = await reader.ReadRemainingDataAsync(4);
 
             if (Encoding.UTF8.GetString(protocolName, 0, protocolName.Length) != "MQTT")
@@ -285,7 +270,12 @@ namespace MQTTnet.Core.Serializer
 
             var connectFlagsReader = new ByteReader(connectFlags);
             connectFlagsReader.Read(); // Reserved.
-            packet.CleanSession = connectFlagsReader.Read();
+
+            var packet = new MqttConnectPacket
+            {
+                CleanSession = connectFlagsReader.Read()
+            };
+
             var willFlag = connectFlagsReader.Read();
             var willQoS = connectFlagsReader.Read(2);
             var willRetain = connectFlagsReader.Read();
@@ -318,7 +308,7 @@ namespace MQTTnet.Core.Serializer
             return packet;
         }
 
-        private async Task<MqttBasePacket> DeserializeSubAck(MqttPacketReader reader)
+        private static async Task<MqttBasePacket> DeserializeSubAck(MqttPacketReader reader)
         {
             var packet = new MqttSubAckPacket
             {
@@ -333,7 +323,7 @@ namespace MQTTnet.Core.Serializer
             return packet;
         }
 
-        private async Task<MqttBasePacket> DeserializeConnAck(MqttPacketReader reader)
+        private static async Task<MqttBasePacket> DeserializeConnAck(MqttPacketReader reader)
         {
             var variableHeader1 = await reader.ReadRemainingDataByteAsync();
             var variableHeader2 = await reader.ReadRemainingDataByteAsync();
@@ -347,7 +337,7 @@ namespace MQTTnet.Core.Serializer
             return packet;
         }
 
-        private void ValidateConnectPacket(MqttConnectPacket packet)
+        private static void ValidateConnectPacket(MqttConnectPacket packet)
         {
             if (string.IsNullOrEmpty(packet.ClientId) && !packet.CleanSession)
             {
@@ -355,7 +345,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private void ValidatePublishPacket(MqttPublishPacket packet)
+        private static void ValidatePublishPacket(MqttPublishPacket packet)
         {
             if (packet.QualityOfServiceLevel == 0 && packet.Dup)
             {
@@ -363,9 +353,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private static readonly byte[] MqttPrefix = Encoding.UTF8.GetBytes("MQTT");
-
-        private Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
         {
             ValidateConnectPacket(packet);
 
@@ -373,7 +361,7 @@ namespace MQTTnet.Core.Serializer
             {
                 // Write variable header
                 output.Write(0x00, 0x04); // 3.1.2.1 Protocol Name
-                output.Write(MqttPrefix);
+                output.Write(MqttV311Prefix);
                 output.Write(0x04); // 3.1.2.2 Protocol Level
 
                 var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags
@@ -420,7 +408,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -435,22 +423,22 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination)
         {
             return SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, destination);
         }
 
-        private Task SerializeAsync(MqttPingReqPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttPingReqPacket packet, IMqttCommunicationChannel destination)
         {
             return SerializeEmptyPacketAsync(MqttControlPacketType.PingReq, destination);
         }
 
-        private Task SerializeAsync(MqttPingRespPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttPingRespPacket packet, IMqttCommunicationChannel destination)
         {
             return SerializeEmptyPacketAsync(MqttControlPacketType.PingResp, destination);
         }
 
-        private Task SerializeAsync(MqttPublishPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttPublishPacket packet, IMqttCommunicationChannel destination)
         {
             ValidatePublishPacket(packet);
 
@@ -485,7 +473,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttPubAckPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttPubAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -496,7 +484,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttPubRecPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttPubRecPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -507,7 +495,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination)
+        private static async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -518,7 +506,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -529,7 +517,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttSubscribePacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttSubscribePacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -549,7 +537,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttSubAckPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttSubAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -568,7 +556,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttUnsubscribePacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttUnsubscribePacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -587,7 +575,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeAsync(MqttUnsubAckPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttUnsubAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
@@ -598,7 +586,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private Task SerializeEmptyPacketAsync(MqttControlPacketType type, IMqttCommunicationChannel destination)
+        private static Task SerializeEmptyPacketAsync(MqttControlPacketType type, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {

MQTTnet.Core/Server/IMqttServer.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Collections.Generic;
+using MQTTnet.Core.Adapter;
+
+namespace MQTTnet.Core.Server
+{
+    public interface IMqttServer
+    {
+        event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+        event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+
+        IList<string> GetConnectedClients();
+        void InjectClient(string identifier, IMqttCommunicationAdapter adapter);
+        void Publish(MqttApplicationMessage applicationMessage);
+        void Start();
+        void Stop();
+    }
+}

MQTTnet.Core/Server/MqttClientMessageQueue.cs

@@ -35,7 +35,7 @@ namespace MQTTnet.Core.Server
             _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter));
             _cancellationTokenSource = new CancellationTokenSource();
 
-            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token));
+            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
         }
 
         public void Stop()
@@ -45,14 +45,13 @@ namespace MQTTnet.Core.Server
             _cancellationTokenSource = null;
         }
 
-        public void Enqueue(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public void Enqueue(MqttPublishPacket publishPacket)
         {
-            if (senderClientSession == null) throw new ArgumentNullException(nameof(senderClientSession));
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
             lock (_pendingPublishPackets)
             {
-                _pendingPublishPackets.Add(new MqttClientPublishPacketContext(senderClientSession, publishPacket));
+                _pendingPublishPackets.Add(new MqttClientPublishPacketContext(publishPacket));
                 _gate.Set();
             }
         }

MQTTnet.Core/Server/MqttClientPublishPacketContext.cs

@@ -5,14 +5,11 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientPublishPacketContext
     {
-        public MqttClientPublishPacketContext(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public MqttClientPublishPacketContext(MqttPublishPacket publishPacket)
         {
-            SenderClientSession = senderClientSession ?? throw new ArgumentNullException(nameof(senderClientSession));
             PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
         }
 
-        public MqttClientSession SenderClientSession { get; }
-
         public MqttPublishPacket PublishPacket { get; }
 
         public int SendTries { get; set; }

MQTTnet.Core/Server/MqttClientSession.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
@@ -13,13 +13,13 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSession : IDisposable
     {
-        private readonly ConcurrentDictionary<ushort, MqttPublishPacket> _pendingIncomingPublications = new ConcurrentDictionary<ushort, MqttPublishPacket>();
+        private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
 
         private readonly MqttClientSubscriptionsManager _subscriptionsManager = new MqttClientSubscriptionsManager();
         private readonly MqttClientMessageQueue _messageQueue;
         private readonly Action<MqttClientSession, MqttPublishPacket> _publishPacketReceivedCallback;
         private readonly MqttServerOptions _options;
-        
+
         private CancellationTokenSource _cancellationTokenSource;
         private IMqttCommunicationAdapter _adapter;
         private string _identifier;
@@ -79,17 +79,16 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        public void EnqueuePublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public void EnqueuePublishPacket(MqttPublishPacket publishPacket)
         {
-            if (senderClientSession == null) throw new ArgumentNullException(nameof(senderClientSession));
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            if (!_subscriptionsManager.IsTopicSubscribed(publishPacket))
+            if (!_subscriptionsManager.IsSubscribed(publishPacket))
             {
                 return;
             }
 
-            _messageQueue.Enqueue(senderClientSession, publishPacket);
+            _messageQueue.Enqueue(publishPacket);
             MqttTrace.Verbose(nameof(MqttClientSession), $"Client '{_identifier}: Enqueued pending publish packet.");
         }
 
@@ -101,34 +100,35 @@ namespace MQTTnet.Core.Server
 
         private Task HandleIncomingPacketAsync(MqttBasePacket packet)
         {
-            var subscribePacket = packet as MqttSubscribePacket;
-            if (subscribePacket != null)
+            if (packet is MqttSubscribePacket subscribePacket)
             {
                 return _adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
             }
 
-            var unsubscribePacket = packet as MqttUnsubscribePacket;
-            if (unsubscribePacket != null)
+            if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
                 return _adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
             }
 
-            var publishPacket = packet as MqttPublishPacket;
-            if (publishPacket != null)
+            if (packet is MqttPublishPacket publishPacket)
             {
                 return HandleIncomingPublishPacketAsync(publishPacket);
             }
 
-            var pubRelPacket = packet as MqttPubRelPacket;
-            if (pubRelPacket != null)
+            if (packet is MqttPubRelPacket pubRelPacket)
             {
                 return HandleIncomingPubRelPacketAsync(pubRelPacket);
             }
 
-            var pubAckPacket = packet as MqttPubAckPacket;
-            if (pubAckPacket != null)
+            if (packet is MqttPubRecPacket pubRecPacket)
             {
-                return HandleIncomingPubAckPacketAsync(pubAckPacket);
+                return _adapter.SendPacketAsync(pubRecPacket.CreateResponse<MqttPubRelPacket>(), _options.DefaultCommunicationTimeout);
+            }
+
+            if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
+            {
+                // Discard message.
+                return Task.FromResult((object)null);
             }
 
             if (packet is MqttPingReqPacket)
@@ -148,39 +148,44 @@ namespace MQTTnet.Core.Server
             return Task.FromResult((object)null);
         }
 
-        private async Task HandleIncomingPubAckPacketAsync(MqttPubAckPacket pubAckPacket)
-        {
-            await Task.FromResult((object)null);
-        }
-
-        private async Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
+        private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
         {
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
                 _publishPacketReceivedCallback(this, publishPacket);
+                return Task.FromResult(0);
             }
-            else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+
+            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
-                await _adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
                 _publishPacketReceivedCallback(this, publishPacket);
+                return _adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
-            else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+
+            if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
             {
-                _pendingIncomingPublications[publishPacket.PacketIdentifier] = publishPacket;
-                await _adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
+                // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
+                lock (_unacknowledgedPublishPackets)
+                {
+                    _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
+                }
+
+                _publishPacketReceivedCallback(this, publishPacket);
+
+                return _adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
+
+            throw new MqttCommunicationException("Received a not supported QoS level.");
         }
 
-        private async Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket)
+        private Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket)
         {
-            MqttPublishPacket publishPacket;
-            if (!_pendingIncomingPublications.TryRemove(pubRelPacket.PacketIdentifier, out publishPacket))
+            lock (_unacknowledgedPublishPackets)
             {
-                return;
+                _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
             }
 
-            await _adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
-            _publishPacketReceivedCallback(this, publishPacket);
+            return _adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
         }
     }
 }

MQTTnet.Core/Server/MqttClientSessionsManager.cs

@@ -22,7 +22,7 @@ namespace MQTTnet.Core.Server
             _options = options ?? throw new ArgumentNullException(nameof(options));
         }
 
-        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; 
+        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
         public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
         {
@@ -127,14 +127,24 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        private void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
         {
-            var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession.ClientId, publishPacket.ToApplicationMessage());
-            ApplicationMessageReceived?.Invoke(this, eventArgs);
-
-            foreach (var clientSession in _clientSessions.Values.ToList())
+            try
             {
-                clientSession.EnqueuePublishPacket(senderClientSession, publishPacket);
+                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, publishPacket.ToApplicationMessage());
+                ApplicationMessageReceived?.Invoke(this, eventArgs);
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClientSessionsManager), exception, "Error while processing application message");
+            }
+
+            lock (_syncRoot)
+            {
+                foreach (var clientSession in _clientSessions.Values.ToList())
+                {
+                    clientSession.EnqueuePublishPacket(publishPacket);
+                }
             }
         }
     }

MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Collections.Concurrent;
+using System.Collections.Generic;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;
 
@@ -7,17 +7,21 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSubscriptionsManager
     {
-        private readonly ConcurrentDictionary<string, MqttQualityOfServiceLevel> _subscribedTopics = new ConcurrentDictionary<string, MqttQualityOfServiceLevel>();
+        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscribedTopics = new Dictionary<string, MqttQualityOfServiceLevel>();
 
         public MqttSubAckPacket Subscribe(MqttSubscribePacket subscribePacket)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
 
             var responsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>();
-            foreach (var topicFilter in subscribePacket.TopicFilters)
+
+            lock (_subscribedTopics)
             {
-                _subscribedTopics[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
-                responsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1); // TODO: Add support for QoS 2.
+                foreach (var topicFilter in subscribePacket.TopicFilters)
+                {
+                    _subscribedTopics[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                    responsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1); // TODO: Add support for QoS 2.
+                }
             }
 
             return responsePacket;
@@ -27,32 +31,37 @@ namespace MQTTnet.Core.Server
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
 
-            foreach (var topicFilter in unsubscribePacket.TopicFilters)
+            lock (_subscribedTopics)
             {
-                MqttQualityOfServiceLevel _;
-                _subscribedTopics.TryRemove(topicFilter, out _);
+                foreach (var topicFilter in unsubscribePacket.TopicFilters)
+                {
+                    _subscribedTopics.Remove(topicFilter);
+                }
             }
 
             return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>();
         }
 
-        public bool IsTopicSubscribed(MqttPublishPacket publishPacket)
+        public bool IsSubscribed(MqttPublishPacket publishPacket)
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            foreach (var subscribedTopic in _subscribedTopics)
+            lock (_subscribedTopics)
             {
-                if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscribedTopic.Key))
+                foreach (var subscribedTopic in _subscribedTopics)
                 {
-                    continue;
-                }
+                    if (publishPacket.QualityOfServiceLevel > subscribedTopic.Value)
+                    {
+                        continue;
+                    }
 
-                if (subscribedTopic.Value < publishPacket.QualityOfServiceLevel)
-                {
-                    continue;
-                }
+                    if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscribedTopic.Key))
+                    {
+                        continue;
+                    }
 
-                return true;
+                    return true;
+                }
             }
 
             return false;

MQTTnet.Core/Server/MqttServer.cs

@@ -4,10 +4,11 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.Internal;
 
 namespace MQTTnet.Core.Server
 {
-    public sealed class MqttServer
+    public sealed class MqttServer : IMqttServer
     {
         private readonly MqttClientSessionsManager _clientSessionsManager;
         private readonly ICollection<IMqttServerAdapter> _adapters;
@@ -33,6 +34,13 @@ namespace MQTTnet.Core.Server
 
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
+        public void Publish(MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+
+            _clientSessionsManager.DispatchPublishPacket(null, applicationMessage.ToPublishPacket());
+        }
+
         public void InjectClient(string identifier, IMqttCommunicationAdapter adapter)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));

MQTTnet.Core/Server/MqttServerDefaultEndpointOptions.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Server
 {
-    public sealed class DefaultEndpointOptions
+    public sealed class MqttServerDefaultEndpointOptions
     {
         public bool IsEnabled { get; set; } = true;
 

MQTTnet.Core/Server/MqttServerOptions.cs

@@ -6,13 +6,13 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttServerOptions
     {
-        public DefaultEndpointOptions DefaultEndpointOptions { get; } = new DefaultEndpointOptions();
+        public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions();
 
         public MqttServerTlsEndpointOptions TlsEndpointOptions { get; } = new MqttServerTlsEndpointOptions();
         
         public int ConnectionBacklog { get; set; } = 10;
 
-        public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
+        public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
 
         public Func<MqttConnectPacket, MqttConnectReturnCode> ConnectionValidator { get; set; }
     }

MQTTnet.Core/Server/MqttServerTlsEndpointOptionsExtensions.cs

@@ -1,19 +0,0 @@
-using System;
-
-namespace MQTTnet.Core.Server
-{
-    public static class MqttServerTlsEndpointOptionsExtensions
-    {
-        public static int GetPort(this DefaultEndpointOptions options)
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
-
-            if (!options.Port.HasValue)
-            {
-                return 1883;
-            }
-
-            return options.Port.Value;
-        }
-    }
-}

README.md

@@ -2,7 +2,7 @@
 <img src="https://github.com/chkr1011/MQTTnet/blob/master/Images/Logo_128x128.png?raw=true" width="128">
 </p>
 
-[![Build Status](https://img.shields.io/nuget/v/MQTTnet.svg)](https://www.nuget.org/packages/MQTTnet/)
+[![NuGet Badge](https://buildstats.info/nuget/MQTTnet)](https://www.nuget.org/packages/MQTTnet)
 
 # MQTTnet
 MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server. The implementation is based on the documentation from http://mqtt.org/.
@@ -12,12 +12,15 @@ MQTTnet is a .NET library for MQTT based communication. It provides a MQTT clien
 * MQTT server (broker) included
 * TLS 1.2 support for client and server (but not UWP servers)
 * Async support
+* Rx support (via another project)
 * List of connected clients available (server only)
 * Extensible communication channels (i.e. In-Memory, TCP, TCP+SSL, WebSockets (not included in this project))
+* Server is able to publish its own messages (no loopback client required)
 * Access to internal trace messages
 * Extensible client credential validation (server only)
-* Unit tested (48+ tests)
+* Unit tested (50+ tests)
 * Lightweight (only the low level implementation of MQTT, no overhead)
+* Interfaces included for mocking and testing
 
 ## Supported frameworks
 * .NET Standard 1.3+
@@ -38,6 +41,7 @@ If you want to contribute to this project just create a pull request.
 ## References
 This library is used in the following projects:
 
+* MQTT Client Rx (Wrapper for Reactive Extensions, https://github.com/1iveowl/MQTTClient.rx)
 * HA4IoT (Open Source Home Automation system for .NET, https://github.com/chkr1011/HA4IoT)
 
 If you use this library and want to see your project here please let me know.

Tests/MQTTnet.Core.Tests/DefaultMqttV311PacketSerializerTests.cs

@@ -405,7 +405,7 @@ namespace MQTTnet.Core.Tests
 
         private void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
         {
-            var serializer = new DefaultMqttV311PacketSerializer();
+            var serializer = new MqttV311PacketSerializer();
             var channel = new TestChannel();
             serializer.SerializeAsync(packet, channel).Wait();
             var buffer = channel.ToArray();
@@ -415,7 +415,7 @@ namespace MQTTnet.Core.Tests
 
         private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
         {
-            var serializer = new DefaultMqttV311PacketSerializer();
+            var serializer = new MqttV311PacketSerializer();
 
             var channel1 = new TestChannel();
             serializer.SerializeAsync(packet, channel1).Wait();

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -68,7 +68,64 @@ namespace MQTTnet.Core.Tests
             Assert.AreEqual(1, receivedMessagesCount);
         }
 
-        private MqttClient ConnectTestClient(string clientId, MqttApplicationMessage willMessage, MqttServer server)
+        [TestMethod]
+        public async Task MqttServer_Unsubscribe()
+        {
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() });
+            s.Start();
+
+            var c1 = ConnectTestClient("c1", null, s);
+            var c2 = ConnectTestClient("c2", null, s);
+
+            var receivedMessagesCount = 0;
+            c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
+            
+            var message = new MqttApplicationMessage("a", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, false);
+
+            await c2.PublishAsync(message);
+            Assert.AreEqual(0, receivedMessagesCount);
+
+            await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
+            await c2.PublishAsync(message);
+
+            await Task.Delay(500);
+            Assert.AreEqual(1, receivedMessagesCount);
+
+            await c1.Unsubscribe("a");
+            await c2.PublishAsync(message);
+
+            await Task.Delay(500);
+            Assert.AreEqual(1, receivedMessagesCount);
+
+            s.Stop();
+            await Task.Delay(500);
+
+            Assert.AreEqual(1, receivedMessagesCount);
+        }
+
+        [TestMethod]
+        public async Task MqttServer_Publish()
+        {
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() });
+            s.Start();
+
+            var c1 = ConnectTestClient("c1", null, s);
+
+            var receivedMessagesCount = 0;
+            c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
+
+            var message = new MqttApplicationMessage("a", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, false);
+            await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
+
+            s.Publish(message);
+            await Task.Delay(500);
+
+            s.Stop();
+
+            Assert.AreEqual(1, receivedMessagesCount);
+        }
+
+        private static MqttClient ConnectTestClient(string clientId, MqttApplicationMessage willMessage, MqttServer server)
         {
             var adapterA = new TestMqttCommunicationAdapter();
             var adapterB = new TestMqttCommunicationAdapter();

Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs

@@ -24,7 +24,7 @@ namespace MQTTnet.Core.Tests
                 QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
             };
 
-            Assert.IsTrue(sm.IsTopicSubscribed(pp));
+            Assert.IsTrue(sm.IsSubscribed(pp));
         }
 
         [TestMethod]
@@ -43,7 +43,7 @@ namespace MQTTnet.Core.Tests
                 QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
             };
 
-            Assert.IsFalse(sm.IsTopicSubscribed(pp));
+            Assert.IsFalse(sm.IsSubscribed(pp));
         }
 
         [TestMethod]
@@ -62,13 +62,13 @@ namespace MQTTnet.Core.Tests
                 QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
             };
 
-            Assert.IsTrue(sm.IsTopicSubscribed(pp));
+            Assert.IsTrue(sm.IsSubscribed(pp));
 
             var up = new MqttUnsubscribePacket();
             up.TopicFilters.Add("A/B/C");
             sm.Unsubscribe(up);
 
-            Assert.IsFalse(sm.IsTopicSubscribed(pp));
+            Assert.IsFalse(sm.IsSubscribed(pp));
         }
     }
 }

Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.TestApp.UniversalWindows
 {
     public sealed partial class MainPage
     {
-        private MqttClient _mqttClient;
+        private IMqttClient _mqttClient;
 
         public MainPage()
         {