Merge branch 'master' of https://github.com/chkr1011/MQTTnet

Fix infinite loop when the client closes the connection

Build/MQTTnet.AspNetCore.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet.AspNetCore</id>
-    <version>2.6.0</version>
+    <version>2.7.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,13 +10,13 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>This is a support library to integrate MQTTnet into AspNetCore.</description>
-    <releaseNotes>* Updated to MQTTnet 2.6.0.
+    <releaseNotes>* Updated to MQTTnet 2.7.0.
     </releaseNotes>
-    <copyright>Copyright Christian Kratky 2016-2017</copyright>
+    <copyright>Copyright Christian Kratky 2016-2018</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
     <dependencies>
       <group targetFramework="netstandard2.0">
-        <dependency id="MQTTnet" version="2.6.0" />
+        <dependency id="MQTTnet" version="2.7.0" />
       </group>
     </dependencies>
   </metadata>

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.6.0</version>
+    <version>2.7.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,28 +10,18 @@
     <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
-    <releaseNotes>* [Core] Merged projects (BREAKING CHANGE! But only namespace changes).
+    <releaseNotes>* [Core] Fixed some still thread blocking parts in the code (thanks to @kpreisser).
-* [Core] Added a strong name for the assembly.
+* [Core] Updated 3rd-Party packages.
-* [Core] Performance optimizations.
+* [Core] Fixed wrong packet identifier calculation (thanks to @benpittoors).
-* [Core] Fixed a logging issue when dealing with IOExceptions.
+* [Core] Fixed an issue when reading the body of a package from a disconnected sender (thanks to @kpreisser).
-* [Core] Fixed a typo in the global logger class (BREAKING CHANGE! Please find new example in Wiki).
+* [Core] Fixed wrong parsing of the body length (thanks to @kpreisser).
-* [Core] Added support for project compilation under macOS (Thanks to @FourOnes).
+* [Client] The client interfaces are now implementing _IDisposable_.
-* [Client] The certificate for encryption is now loaded as a _X509Certificate2_ which prodives more data (Thanks to @cvellan).
+* [Client] The disconnected event now contains the exception which was thrown and causing the disconnect.
-* [Client] Fixed an issue in _ManagedClient_ which can cause the client to stop when publishing subscriptions.
+* [Server] Fixed an issue which lets the server block 1 second after accepting a new connection (thanks to @kpreisser).
-* [Client] Fixed an issue in _ManagedClient_ which prevents changing the QoS of an existing subscription (BREAKING CHANGE!).
+* [Server] The server now allows managing client subscriptions.
-* [Client] Fixed an issue in _MqttClientOptionsBuilder_ which prevents adding TLS options to the client options when calling _Build()_. (Thanks to @cvellan).
+* [Server] Added events for topic subscriptions.
-* [Client] Added an overload for subscribing at both clients which reduces required code.
-* [Client] Added the _ClientId_ to the application message interceptor context.
-* [Server] The application message interceptor can now delete any received application message.
-* [Server] Added a ConnectionValidator context to align with other APIs (BREAKING CHANGE! Please find new example in Wiki).
-* [Server] Added an interface for the _MqttServerOptions_.
-* [Server] Added packet statistics for the connected clients.
-* [Server] Fixed a security issue which sends retained packages to a failed subscription.
-* [Server] Fixed the response (MaximumQoS) of a subscription (Thanks to @redbeans2017).
-* [Server] The keep alive timeouts are now checked for every client (Thanks to @RainerMueller82).
-* [Server] The used IP addresses can be specified now (IPAddress.Any as default).
     </releaseNotes>
-    <copyright>Copyright Christian Kratky 2016-2017</copyright>
+    <copyright>Copyright Christian Kratky 2016-2018</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
     <dependencies>
 
@@ -75,7 +65,6 @@
 
     <!-- .NET Framework -->
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net452\MQTTnet.*" target="lib\net452\"/>
-
     <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\net461\MQTTnet.*" target="lib\net461\"/>
     
   </files>

Documents/Import_CodeSigningKey.md

@@ -0,0 +1,6 @@
+# Import codeSigningKey.pfx
+In order to import the key for code signing on a new developer machine use the following command within the VS developer command line:
+
+> sn –i codeSigningKey.pfx VS_KEY_EFCA4C5B6DFD4B4F
+
+The container name may be different.

Frameworks/MQTTnet.AspnetCore/MQTTnet.AspnetCore.csproj

@@ -16,9 +16,9 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.0" />
+    <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.1" />
-    <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.0" />
+    <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.1" />
-    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.1" />
   </ItemGroup>
 
   <ItemGroup>

Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs

@@ -17,8 +17,7 @@ namespace MQTTnet.Adapter
     public class MqttChannelAdapter : IMqttChannelAdapter
     {
         private const uint ErrorOperationAborted = 0x800703E3;
-
+        private const int ReadBufferSize = 4096;  // TODO: Move buffer size to config
-        private static readonly byte[] EmptyBody = new byte[0];
 
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
         private readonly IMqttNetLogger _logger;
@@ -91,28 +90,35 @@ namespace MQTTnet.Adapter
             MqttBasePacket packet = null;
             await ExecuteAndWrapExceptionAsync(async () =>
             {
-                ReceivedMqttPacket receivedMqttPacket;
+                ReceivedMqttPacket receivedMqttPacket = null;
-                if (timeout > TimeSpan.Zero)
+                try
                 {
-                    receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
+                    if (timeout > TimeSpan.Zero)
-                }
+                    {
-                else
+                        receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
-                {
+                    }
-                    receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
+                    else
-                }
+                    {
+                        receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, cancellationToken).ConfigureAwait(false);
+                    }
 
-                if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
+                    if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested)
-                {
+                    {
-                    throw new TaskCanceledException();
+                        throw new TaskCanceledException();
-                }
+                    }
 
-                packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
+                    packet = PacketSerializer.Deserialize(receivedMqttPacket.Header, receivedMqttPacket.Body);
-                if (packet == null)
+                    if (packet == null)
-                {
+                    {
-                    throw new MqttProtocolViolationException("Received malformed packet.");
+                        throw new MqttProtocolViolationException("Received malformed packet.");
-                }
+                    }
 
-                _logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
+                    _logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
+                }
+                finally
+                {
+                    receivedMqttPacket?.Dispose();
+                }
             }).ConfigureAwait(false);
 
             return packet;
@@ -120,7 +126,7 @@ namespace MQTTnet.Adapter
 
         private static async Task<ReceivedMqttPacket> ReceiveAsync(Stream stream, CancellationToken cancellationToken)
         {
-            var header = MqttPacketReader.ReadHeaderFromSource(stream, cancellationToken);
+            var header = await MqttPacketReader.ReadHeaderAsync(stream, cancellationToken).ConfigureAwait(false);
             if (header == null)
             {
                 return null;
@@ -128,17 +134,34 @@ namespace MQTTnet.Adapter
 
             if (header.BodyLength == 0)
             {
-                return new ReceivedMqttPacket(header, EmptyBody);
+                return new ReceivedMqttPacket(header, new MemoryStream(new byte[0], false));
             }
 
-            var body = new byte[header.BodyLength];
+            var body = header.BodyLength <= ReadBufferSize ? new MemoryStream(header.BodyLength) : new MemoryStream();
-
+            
-            var offset = 0;
+            var buffer = new byte[ReadBufferSize];
-            do
+            while (body.Length < header.BodyLength)
             {
-                var readBytesCount = await stream.ReadAsync(body, offset, body.Length - offset, cancellationToken).ConfigureAwait(false);
+                var bytesLeft = header.BodyLength - (int)body.Length;
-                offset += readBytesCount;
+                if (bytesLeft > buffer.Length)
-            } while (offset < header.BodyLength);
+                {
+                    bytesLeft = buffer.Length;
+                }
+
+                var readBytesCount = await stream.ReadAsync(buffer, 0, bytesLeft, cancellationToken).ConfigureAwait(false);
+                
+                // Check if the client closed the connection before sending the full body.
+                if (readBytesCount == 0)
+                {
+                    throw new MqttCommunicationException("Connection closed while reading remaining packet body.");
+                }
+
+                // Here is no need to await because internally only an array is used and no real I/O operation is made.
+                // Using async here will only generate overhead.
+                body.Write(buffer, 0, readBytesCount);
+            }
+
+            body.Seek(0L, SeekOrigin.Begin);
             
             return new ReceivedMqttPacket(header, body);
         }

Frameworks/MQTTnet.NetStandard/Adapter/ReceivedMqttPacket.cs

@@ -1,11 +1,12 @@
 using System;
+using System.IO;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Adapter
 {
-    public class ReceivedMqttPacket
+    public sealed class ReceivedMqttPacket : IDisposable
     {
-        public ReceivedMqttPacket(MqttPacketHeader header, byte[] body)
+        public ReceivedMqttPacket(MqttPacketHeader header, MemoryStream body)
         {
             Header = header ?? throw new ArgumentNullException(nameof(header));
             Body = body ?? throw new ArgumentNullException(nameof(body));
@@ -13,6 +14,11 @@ namespace MQTTnet.Adapter
 
         public MqttPacketHeader Header { get; }
 
-        public byte[] Body { get; }
+        public MemoryStream Body { get; }
+
+        public void Dispose()
+        {
+            Body?.Dispose();
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/Client/IMqttClient.cs

@@ -4,7 +4,7 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.Client
 {
-    public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
+    public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
     {
         bool IsConnected { get; }
 

Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs

@@ -12,15 +12,15 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Client
 {
-    public class MqttClient : IMqttClient, IDisposable
+    public class MqttClient : IMqttClient
     {
+        private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         private readonly IMqttClientAdapterFactory _adapterFactory;
         private readonly MqttPacketDispatcher _packetDispatcher;
         private readonly IMqttNetLogger _logger;
 
         private IMqttClientOptions _options;
         private bool _isReceivingPackets;
-        private int _latestPacketIdentifier;
         private CancellationTokenSource _cancellationTokenSource;
         private IMqttChannelAdapter _adapter;
 
@@ -49,7 +49,7 @@ namespace MQTTnet.Client
             {
                 _options = options;
                 _cancellationTokenSource = new CancellationTokenSource();
-                _latestPacketIdentifier = 0;
+                _packetIdentifierProvider.Reset();
                 _packetDispatcher.Reset();
 
                 _adapter = _adapterFactory.CreateClientAdapter(options.ChannelOptions, _logger);
@@ -75,7 +75,8 @@ namespace MQTTnet.Client
             catch (Exception exception)
             {
                 _logger.Error<MqttClient>(exception, "Error while connecting with server.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
+
                 throw;
             }
         }
@@ -93,7 +94,7 @@ namespace MQTTnet.Client
             }
             finally
             {
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(null).ConfigureAwait(false);
             }
         }
 
@@ -105,7 +106,7 @@ namespace MQTTnet.Client
 
             var subscribePacket = new MqttSubscribePacket
             {
-                PacketIdentifier = GetNewPacketIdentifier(),
+                PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(),
                 TopicFilters = topicFilters.ToList()
             };
 
@@ -127,7 +128,7 @@ namespace MQTTnet.Client
 
             var unsubscribePacket = new MqttUnsubscribePacket
             {
-                PacketIdentifier = GetNewPacketIdentifier(),
+                PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(),
                 TopicFilters = topicFilters.ToList()
             };
 
@@ -155,7 +156,7 @@ namespace MQTTnet.Client
                         {
                             foreach (var publishPacket in qosGroup)
                             {
-                                publishPacket.PacketIdentifier = GetNewPacketIdentifier();
+                                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
                                 await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket).ConfigureAwait(false);
                             }
 
@@ -165,7 +166,7 @@ namespace MQTTnet.Client
                         {
                             foreach (var publishPacket in qosGroup)
                             {
-                                publishPacket.PacketIdentifier = GetNewPacketIdentifier();
+                                publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier();
                                 var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket).ConfigureAwait(false);
                                 await SendAndReceiveAsync<MqttPubCompPacket>(pubRecPacket.CreateResponse<MqttPubRelPacket>()).ConfigureAwait(false);
                             }
@@ -216,7 +217,7 @@ namespace MQTTnet.Client
             if (IsConnected) throw new MqttProtocolViolationException(message);
         }
 
-        private async Task DisconnectInternalAsync()
+        private async Task DisconnectInternalAsync(Exception exception)
         {
             var clientWasConnected = IsConnected;
             IsConnected = false;
@@ -236,14 +237,14 @@ namespace MQTTnet.Client
                 await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
                 _logger.Info<MqttClient>("Disconnected from adapter.");
             }
-            catch (Exception exception)
+            catch (Exception adapterException)
             {
-                _logger.Warning<MqttClient>(exception, "Error while disconnecting from adapter.");
+                _logger.Warning<MqttClient>(adapterException, "Error while disconnecting from adapter.");
             }
             finally
             {
                 _logger.Info<MqttClient>("Disconnected.");
-                Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected));
+                Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected, exception));
             }
         }
 
@@ -339,11 +340,6 @@ namespace MQTTnet.Client
             return (TResponsePacket)await packetAwaiter.ConfigureAwait(false);
         }
 
-        private ushort GetNewPacketIdentifier()
-        {
-            return (ushort)Interlocked.Increment(ref _latestPacketIdentifier);
-        }
-
         private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
         {
             _logger.Info<MqttClient>("Start sending keep alive packets.");
@@ -363,7 +359,7 @@ namespace MQTTnet.Client
                     return;
                 }
 
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(null).ConfigureAwait(false);
             }
             catch (MqttCommunicationException exception)
             {
@@ -373,12 +369,12 @@ namespace MQTTnet.Client
                 }
 
                 _logger.Warning<MqttClient>(exception, "MQTT communication exception while sending/receiving keep alive packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
                 _logger.Warning<MqttClient>(exception, "Unhandled exception while sending/receiving keep alive packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             finally
             {
@@ -413,7 +409,7 @@ namespace MQTTnet.Client
                     return;
                 }
 
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(null).ConfigureAwait(false);
             }
             catch (MqttCommunicationException exception)
             {
@@ -423,12 +419,12 @@ namespace MQTTnet.Client
                 }
 
                 _logger.Warning<MqttClient>(exception, "MQTT communication exception while receiving packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
                 _logger.Error<MqttClient>(exception, "Unhandled exception while receiving packets.");
-                await DisconnectInternalAsync().ConfigureAwait(false);
+                await DisconnectInternalAsync(exception).ConfigureAwait(false);
             }
             finally
             {

Frameworks/MQTTnet.NetStandard/Client/MqttClientDisconnectedEventArgs.cs

@@ -4,11 +4,14 @@ namespace MQTTnet.Client
 {
     public class MqttClientDisconnectedEventArgs : EventArgs
     {
-        public MqttClientDisconnectedEventArgs(bool clientWasConnected)
+        public MqttClientDisconnectedEventArgs(bool clientWasConnected, Exception exception)
         {
             ClientWasConnected = clientWasConnected;
+            Exception = exception;
         }
 
         public bool ClientWasConnected { get; }
+
+        public Exception Exception { get; }
     }
 }

Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs

@@ -83,7 +83,7 @@ namespace MQTTnet.Client
             {
                 _packetByResponseType[responseType] = tcs;
             }
-
+            
             return tcs;
         }
 

Frameworks/MQTTnet.NetStandard/Client/MqttPacketIdentifierProvider.cs

@@ -0,0 +1,32 @@
+namespace MQTTnet.Client
+{
+    public class MqttPacketIdentifierProvider
+    {
+        private readonly object _syncRoot = new object();
+        private ushort _value;
+
+        public void Reset()
+        {
+            lock (_syncRoot)
+            {
+                _value = 0;
+            }
+        }
+
+        public ushort GetNewPacketIdentifier()
+        {
+            lock (_syncRoot)
+            {
+                _value++;
+
+                if (_value == 0)
+                {
+                    // As per official MQTT documentation the package identifier should never be 0.
+                    _value = 1;
+                }
+
+                return _value;
+            }
+        }
+    }
+}

Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs

@@ -110,22 +110,14 @@ namespace MQTTnet.Implementations
                 {
                     // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
                 }
-                catch (SocketException exception)
+                catch (Exception exception)
                 {
-                    if (exception.SocketErrorCode == SocketError.OperationAborted)
+                    if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
                     {
                         return;
                     }
 
                     _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
-                }
-                catch (Exception exception)
-                {
-                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
-                }
-                finally
-                {
-                    //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
             }
@@ -153,22 +145,14 @@ namespace MQTTnet.Implementations
                 {
                     // It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
                 }
-                catch (SocketException exception)
+                catch (Exception exception)
                 {
-                    if (exception.SocketErrorCode == SocketError.OperationAborted)
+                    if (exception is SocketException s && s.SocketErrorCode == SocketError.OperationAborted)
                     {
                         return;
                     }
 
-                    _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at default endpoint.");
-                }
-                catch (Exception exception)
-                {
                     _logger.Error<MqttServerAdapter>(exception, "Error while accepting connection at TLS endpoint.");
-                }
-                finally
-                {
-                    //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
             }

Frameworks/MQTTnet.NetStandard/Internal/TaskExtensions.cs

@@ -30,7 +30,7 @@ namespace MQTTnet.Internal
 
                     if (task.IsFaulted)
                     {
-                        throw new MqttCommunicationException(task.Exception.GetBaseException());
+                        throw new MqttCommunicationException(task.Exception?.GetBaseException());
                     }
                 }
                 finally

Frameworks/MQTTnet.NetStandard/ManagedClient/IManagedMqttClient.cs

@@ -5,7 +5,7 @@ using MQTTnet.Client;
 
 namespace MQTTnet.ManagedClient
 {
-    public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher
+    public interface IManagedMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
     {
         bool IsConnected { get; }
 

Frameworks/MQTTnet.NetStandard/ManagedClient/ManagedMqttClient.cs

@@ -11,7 +11,7 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.ManagedClient
 {
-    public class ManagedMqttClient : IManagedMqttClient, IDisposable
+    public class ManagedMqttClient : IManagedMqttClient
     {
         private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();

Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Serializer
@@ -10,6 +11,6 @@ namespace MQTTnet.Serializer
 
         ICollection<ArraySegment<byte>> Serialize(MqttBasePacket mqttPacket);
 
-        MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body);
+        MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body);
     }
 }

Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs

@@ -22,17 +22,21 @@ namespace MQTTnet.Serializer
 
         public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength;
 
-        public static MqttPacketHeader ReadHeaderFromSource(Stream stream, CancellationToken cancellationToken)
+        public static async Task<MqttPacketHeader> ReadHeaderAsync(Stream stream, CancellationToken cancellationToken)
         {
-            var buffer = stream.ReadByte();
+            // Wait for the next package which starts with the header. At this point there will probably
-            if (buffer == -1)
+            // some large delay and thus the thread should be put back to the pool (await). So ReadByte()
+            // is not an option here.
+            var buffer = new byte[1];
+            var readCount = await stream.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
+            if (readCount <= 0)
             {
                 return null;
             }
-            
+
-            var fixedHeader = (byte)buffer;
+            var fixedHeader = buffer[0];
             var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4);
-            var bodyLength = ReadBodyLengthFromSource(stream, cancellationToken);
+            var bodyLength = await ReadBodyLengthAsync(stream, cancellationToken).ConfigureAwait(false);
 
             return new MqttPacketHeader
             {
@@ -80,13 +84,14 @@ namespace MQTTnet.Serializer
             return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
         }
 
-        private static int ReadBodyLengthFromSource(Stream stream, CancellationToken cancellationToken)
+        private static async Task<int> ReadBodyLengthAsync(Stream stream, CancellationToken cancellationToken)
         {
-            // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
+            // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
             var multiplier = 1;
             var value = 0;
             byte encodedByte;
 
+            var buffer = new byte[1];
             var readBytes = new List<byte>();
             do
             {
@@ -95,21 +100,22 @@ namespace MQTTnet.Serializer
                     throw new TaskCanceledException();
                 }
 
-                var buffer = stream.ReadByte();
+                var readCount = await stream.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false);
-                if (buffer == -1)
+                if (readCount <= 0)
                 {
                     throw new MqttCommunicationException("Connection closed while reading remaining length data.");
                 }
 
-                encodedByte = (byte)buffer;
+                encodedByte = buffer[0];
                 readBytes.Add(encodedByte);
 
                 value += (byte)(encodedByte & 127) * multiplier;
-                multiplier *= 128;
                 if (multiplier > 128 * 128 * 128)
                 {
                     throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
                 }
+
+                multiplier *= 128;
             } while ((encodedByte & 128) != 0);
 
             return value;

Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs

@@ -42,13 +42,12 @@ namespace MQTTnet.Serializer
             }
         }
 
-        public MqttBasePacket Deserialize(MqttPacketHeader header, byte[] body)
+        public MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body)
         {
             if (header == null) throw new ArgumentNullException(nameof(header));
             if (body == null) throw new ArgumentNullException(nameof(body));
 
-            using (var bodyStream = new MemoryStream(body))
+            using (var reader = new MqttPacketReader(header, body))
-            using (var reader = new MqttPacketReader(header, bodyStream))
             {
                 return Deserialize(header, reader);
             }

Frameworks/MQTTnet.NetStandard/Server/ConnectedMqttClient.cs

@@ -12,5 +12,7 @@ namespace MQTTnet.Server
         public TimeSpan LastPacketReceived { get; set; }
 
         public TimeSpan LastNonKeepAlivePacketReceived { get; set; }
+
+        public int PendingApplicationMessages { get; set; }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/IMqttServer.cs

@@ -6,12 +6,18 @@ namespace MQTTnet.Server
 {
     public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher
     {
-        event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
-        event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
         event EventHandler<MqttServerStartedEventArgs> Started;
 
+        event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
+        event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
+        event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
+        
         Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();
 
+        Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters);
+        Task UnsubscribeAsync(string clientId, IList<string> topicFilters);
+
         Task StartAsync(IMqttServerOptions options);
         Task StopAsync();
     }

Frameworks/MQTTnet.NetStandard/Server/IMqttServerOptions.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Net;
 
 namespace MQTTnet.Server
 {

Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs

@@ -12,18 +12,21 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClientPendingMessagesQueue : IDisposable
     {
-        private readonly BlockingCollection<MqttBasePacket> _queue = new BlockingCollection<MqttBasePacket>();
+        private readonly ConcurrentQueue<MqttBasePacket> _queue = new ConcurrentQueue<MqttBasePacket>();
+        private readonly SemaphoreSlim _queueWaitSemaphore = new SemaphoreSlim(0);
         private readonly IMqttServerOptions _options;
-        private readonly MqttClientSession _session;
+        private readonly MqttClientSession _clientSession;
         private readonly IMqttNetLogger _logger;
 
-        public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession session, IMqttNetLogger logger)
+        public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession clientSession, IMqttNetLogger logger)
         {
-            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
-            _session = session ?? throw new ArgumentNullException(nameof(session));
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
+        public int Count => _queue.Count;
+
         public void Start(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -40,8 +43,10 @@ namespace MQTTnet.Server
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
-            _queue.Add(packet);
+            _queue.Enqueue(packet);
-            _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _session.ClientId);
+            _queueWaitSemaphore.Release();
+
+            _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _clientSession.ClientId);
         }
 
         private async Task SendQueuedPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
@@ -58,7 +63,7 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Error<MqttClientPendingMessagesQueue>(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _session.ClientId);
+                _logger.Error<MqttClientPendingMessagesQueue>(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _clientSession.ClientId);
             }
         }
 
@@ -67,27 +72,32 @@ namespace MQTTnet.Server
             MqttBasePacket packet = null;
             try
             {
-                packet = _queue.Take(cancellationToken);
+                await _queueWaitSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+                if (!_queue.TryDequeue(out packet))
+                {
+                    throw new InvalidOperationException(); // should not happen
+                }
+
                 await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
 
-                _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _session.ClientId);
+                _logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId);
             }
             catch (Exception exception)
             {
                 if (exception is MqttCommunicationTimedOutException)
                 {
-                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _session.ClientId);
+                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _clientSession.ClientId);
                 }
                 else if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _session.ClientId);
+                    _logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _clientSession.ClientId);
                 }
                 else if (exception is OperationCanceledException)
                 {
                 }
                 else
                 {
-                    _logger.Error<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed (ClientId: {0}).", _session.ClientId);
+                    _logger.Error<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed (ClientId: {0}).", _clientSession.ClientId);
                 }
 
                 if (packet is MqttPublishPacket publishPacket)
@@ -95,17 +105,18 @@ namespace MQTTnet.Server
                     if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
                     {
                         publishPacket.Dup = true;
-                        _queue.Add(packet, CancellationToken.None);
+                        _queue.Enqueue(packet);
+                        _queueWaitSemaphore.Release();
                     }
                 }
 
-                await _session.StopAsync();
+                await _clientSession.StopAsync();
             }
         }
 
         public void Dispose()
         {
-            _queue?.Dispose();
+            _queueWaitSemaphore?.Dispose();
         }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
@@ -20,10 +21,7 @@ namespace MQTTnet.Server
         private readonly IMqttServerOptions _options;
         private readonly IMqttNetLogger _logger;
 
-        private readonly MqttClientSessionsManager _sessionsManager;
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
-        private readonly MqttClientSubscriptionsManager _subscriptionsManager;
-        private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
 
         private IMqttChannelAdapter _adapter;
         private CancellationTokenSource _cancellationTokenSource;
@@ -33,20 +31,24 @@ namespace MQTTnet.Server
             string clientId,
             IMqttServerOptions options,
             MqttRetainedMessagesManager retainedMessagesManager,
-            MqttClientSessionsManager sessionsManager,
             IMqttNetLogger logger)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
-            _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
             ClientId = clientId;
             
-            _subscriptionsManager = new MqttClientSubscriptionsManager(_options);
+            SubscriptionsManager = new MqttClientSubscriptionsManager(_options, clientId);
-            _pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
+            PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
         }
 
+        public Func<MqttClientSession, MqttApplicationMessage, Task> ApplicationMessageReceivedCallback { get; set; }
+
+        public MqttClientSubscriptionsManager SubscriptionsManager { get; }
+
+        public MqttClientPendingMessagesQueue PendingMessagesQueue { get; }
+
         public string ClientId { get; }
 
         public MqttProtocolVersion? ProtocolVersion => _adapter?.PacketSerializer.ProtocolVersion;
@@ -70,7 +72,7 @@ namespace MQTTnet.Server
                 _adapter = adapter;
                 _cancellationTokenSource = cancellationTokenSource;
 
-                _pendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
+                PendingMessagesQueue.Start(adapter, cancellationTokenSource.Token);
 
                 _lastPacketReceivedTracker.Restart();
                 _lastNonKeepAlivePacketReceivedTracker.Restart();
@@ -122,7 +124,7 @@ namespace MQTTnet.Server
                 if (willMessage != null)
                 {
                     _willMessage = null; //clear willmessage so it is send just once
-                    await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage).ConfigureAwait(false);
+                    await ApplicationMessageReceivedCallback(this, willMessage).ConfigureAwait(false);
                 }
             }
         }
@@ -131,7 +133,7 @@ namespace MQTTnet.Server
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            var result = await _subscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
+            var result = await SubscriptionsManager.CheckSubscriptionsAsync(applicationMessage);
             if (!result.IsSubscribed)
             {
                 return;
@@ -140,12 +142,40 @@ namespace MQTTnet.Server
             var publishPacket = applicationMessage.ToPublishPacket();
             publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
 
-            _pendingMessagesQueue.Enqueue(publishPacket);
+            PendingMessagesQueue.Enqueue(publishPacket);
+        }
+
+        public Task SubscribeAsync(IList<TopicFilter> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            var response = SubscriptionsManager.SubscribeAsync(new MqttSubscribePacket
+            {
+                TopicFilters = topicFilters
+            });
+
+            return response;
+        }
+
+        public Task UnsubscribeAsync(IList<string> topicFilters)
+        {
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            var response = SubscriptionsManager.UnsubscribeAsync(new MqttUnsubscribePacket
+            {
+                TopicFilters = topicFilters
+            });
+
+            return response;
         }
 
         public void Dispose()
         {
-            _pendingMessagesQueue?.Dispose();
+            ApplicationMessageReceivedCallback = null;
+
+            SubscriptionsManager?.Dispose();
+            PendingMessagesQueue?.Dispose();
+
             _cancellationTokenSource?.Dispose();
         }
 
@@ -231,7 +261,7 @@ namespace MQTTnet.Server
 
         private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
-            var subscribeResult = await _subscriptionsManager.SubscribeAsync(subscribePacket, ClientId);
+            var subscribeResult = await SubscriptionsManager.SubscribeAsync(subscribePacket).ConfigureAwait(false);
             await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, subscribeResult.ResponsePacket).ConfigureAwait(false);
 
             if (subscribeResult.CloseConnection)
@@ -245,8 +275,7 @@ namespace MQTTnet.Server
 
         private async Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
         {
-            var unsubscribeResult = await _subscriptionsManager.UnsubscribeAsync(unsubscribePacket);
+            var unsubscribeResult = await SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
-
             await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, unsubscribeResult);
         }
 
@@ -267,7 +296,7 @@ namespace MQTTnet.Server
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
                     {
-                        return _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
+                        return ApplicationMessageReceivedCallback?.Invoke(this, applicationMessage);
                     }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                     {
@@ -286,7 +315,7 @@ namespace MQTTnet.Server
 
         private async Task HandleIncomingPublishPacketWithQoS1(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
-            await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
+            await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
 
             var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier };
             await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);
@@ -295,7 +324,7 @@ namespace MQTTnet.Server
         private async Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
-            await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
+            await ApplicationMessageReceivedCallback(this, applicationMessage).ConfigureAwait(false);
 
             var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier };
             await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, response).ConfigureAwait(false);

Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs

@@ -12,27 +12,32 @@ using MQTTnet.Serializer;
 
 namespace MQTTnet.Server
 {
-    public sealed class MqttClientSessionsManager
+    public sealed class MqttClientSessionsManager : IDisposable
     {
         private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
 
         private readonly IMqttServerOptions _options;
-        private readonly MqttServer _server;
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
         private readonly IMqttNetLogger _logger;
 
-        public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, MqttServer server, IMqttNetLogger logger)
+        public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
         {
-            _server = server ?? throw new ArgumentNullException(nameof(server));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
-        public async Task RunClientSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
+        public Action<ConnectedMqttClient> ClientConnectedCallback { get; set; }
+        public Action<ConnectedMqttClient> ClientDisconnectedCallback { get; set; }
+        public Action<string, TopicFilter> ClientSubscribedTopicCallback { get; set; }
+        public Action<string, string> ClientUnsubscribedTopicCallback { get; set; }
+        public Action<string, MqttApplicationMessage> ApplicationMessageReceivedCallback { get; set; }
+        
+        public async Task RunSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
         {
             var clientId = string.Empty;
+            MqttClientSession clientSession = null;
             try
             {
                 if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false) is MqttConnectPacket connectPacket))
@@ -56,21 +61,22 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var clientSession = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
+                var result = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
+                clientSession = result.Session;
 
                 await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
                 {
                     ConnectReturnCode = connectReturnCode,
-                    IsSessionPresent = clientSession.IsExistingSession
+                    IsSessionPresent = result.IsExistingSession
                 }).ConfigureAwait(false);
 
-                _server.OnClientConnected(new ConnectedMqttClient
+                ClientConnectedCallback?.Invoke(new ConnectedMqttClient
                 {
                     ClientId = clientId,
                     ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
                 });
 
-                await clientSession.Session.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
+                await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -87,10 +93,11 @@ namespace MQTTnet.Server
                     // ignored
                 }
 
-                _server.OnClientDisconnected(new ConnectedMqttClient
+                ClientDisconnectedCallback?.Invoke(new ConnectedMqttClient
                 {
                     ClientId = clientId,
-                    ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
+                    ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion,
+                    PendingApplicationMessages = clientSession?.PendingMessagesQueue.Count ?? 0
                 });
             }
         }
@@ -123,7 +130,8 @@ namespace MQTTnet.Server
                     ClientId = s.Value.ClientId,
                     ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311,
                     LastPacketReceived = s.Value.LastPacketReceived,
-                    LastNonKeepAlivePacketReceived = s.Value.LastNonKeepAlivePacketReceived
+                    LastNonKeepAlivePacketReceived = s.Value.LastNonKeepAlivePacketReceived,
+                    PendingApplicationMessages = s.Value.PendingMessagesQueue.Count
                 }).ToList();
             }
             finally
@@ -147,7 +155,7 @@ namespace MQTTnet.Server
                     await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
                 }
 
-                _server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage);
+                ApplicationMessageReceivedCallback?.Invoke(senderClientSession?.ClientId, applicationMessage);
             }
             catch (Exception exception)
             {
@@ -168,6 +176,48 @@ namespace MQTTnet.Server
             }
         }
 
+        public async Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            await _semaphore.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                if (!_sessions.TryGetValue(clientId, out var session))
+                {
+                    throw new InvalidOperationException($"Client session {clientId} is unknown.");
+                }
+
+                await session.SubscribeAsync(topicFilters);
+            }
+            finally
+            {
+                _semaphore.Release();
+            }
+        }
+
+        public async Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            await _semaphore.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                if (!_sessions.TryGetValue(clientId, out var session))
+                {
+                    throw new InvalidOperationException($"Client session {clientId} is unknown.");
+                }
+
+                await session.UnsubscribeAsync(topicFilters);
+            }
+            finally
+            {
+                _semaphore.Release();
+            }
+        }
+
         private MqttApplicationMessage InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
         {
             if (_options.ApplicationMessageInterceptor == null)
@@ -229,7 +279,14 @@ namespace MQTTnet.Server
                 {
                     isExistingSession = false;
 
-                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, this, _logger);
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, _retainedMessagesManager, _logger)
+                    {
+                        ApplicationMessageReceivedCallback = DispatchApplicationMessageAsync
+                    };
+
+                    clientSession.SubscriptionsManager.TopicSubscribedCallback = ClientSubscribedTopicCallback;
+                    clientSession.SubscriptionsManager.TopicUnsubscribedCallback = ClientUnsubscribedTopicCallback;
+
                     _sessions[connectPacket.ClientId] = clientSession;
 
                     _logger.Trace<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
@@ -242,5 +299,16 @@ namespace MQTTnet.Server
                 _semaphore.Release();
             }
         }
+
+        public void Dispose()
+        {
+            ClientConnectedCallback = null;
+            ClientDisconnectedCallback = null;
+            ClientSubscribedTopicCallback = null;
+            ClientUnsubscribedTopicCallback = null;
+            ApplicationMessageReceivedCallback = null;
+            
+            _semaphore?.Dispose();
+        }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscribedTopicEventArgs.cs

@@ -0,0 +1,17 @@
+using System;
+
+namespace MQTTnet.Server
+{
+    public class MqttClientSubscribedTopicEventArgs : EventArgs
+    {
+        public MqttClientSubscribedTopicEventArgs(string clientId, TopicFilter topicFilter)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
+        }
+
+        public string ClientId { get; }
+
+        public TopicFilter TopicFilter { get; }
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs

@@ -8,18 +8,23 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
-    public sealed class MqttClientSubscriptionsManager
+    public sealed class MqttClientSubscriptionsManager : IDisposable
     {
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly IMqttServerOptions _options;
-
+        private readonly string _clientId;
-        public MqttClientSubscriptionsManager(IMqttServerOptions options)
+        
+        public MqttClientSubscriptionsManager(IMqttServerOptions options, string clientId)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
         }
 
-        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket, string clientId)
+        public Action<string, TopicFilter> TopicSubscribedCallback { get; set; }
+        public Action<string, string> TopicUnsubscribedCallback { get; set; }
+
+        public async Task<MqttClientSubscribeResult> SubscribeAsync(MqttSubscribePacket subscribePacket)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
 
@@ -34,7 +39,7 @@ namespace MQTTnet.Server
             {
                 foreach (var topicFilter in subscribePacket.TopicFilters)
                 {
-                    var interceptorContext = InterceptSubscribe(clientId, topicFilter);
+                    var interceptorContext = InterceptSubscribe(topicFilter);
                     if (!interceptorContext.AcceptSubscription)
                     {
                         result.ResponsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.Failure);
@@ -52,6 +57,7 @@ namespace MQTTnet.Server
                     if (interceptorContext.AcceptSubscription)
                     {
                         _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                        TopicSubscribedCallback?.Invoke(_clientId, topicFilter);
                     }
                 }
             }
@@ -73,6 +79,7 @@ namespace MQTTnet.Server
                 foreach (var topicFilter in unsubscribePacket.TopicFilters)
                 {
                     _subscriptions.Remove(topicFilter);
+                    TopicUnsubscribedCallback?.Invoke(_clientId, topicFilter);
                 }
             }
             finally
@@ -117,9 +124,25 @@ namespace MQTTnet.Server
             }
         }
 
-        private MqttSubscriptionInterceptorContext InterceptSubscribe(string clientId, TopicFilter topicFilter)
+        public void Dispose()
         {
-            var interceptorContext = new MqttSubscriptionInterceptorContext(clientId, topicFilter);
+            _semaphore?.Dispose();
+        }
+
+        private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
+        {
+            switch (qualityOfServiceLevel)
+            {
+                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
+                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
+                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
+                default: return MqttSubscribeReturnCode.Failure;
+            }
+        }
+
+        private MqttSubscriptionInterceptorContext InterceptSubscribe(TopicFilter topicFilter)
+        {
+            var interceptorContext = new MqttSubscriptionInterceptorContext(_clientId, topicFilter);
             _options.SubscriptionInterceptor?.Invoke(interceptorContext);
             return interceptorContext;
         }
@@ -146,16 +169,5 @@ namespace MQTTnet.Server
                 QualityOfServiceLevel = effectiveQoS
             };
         }
-
-        private static MqttSubscribeReturnCode ConvertToMaximumQoS(MqttQualityOfServiceLevel qualityOfServiceLevel)
-        {
-            switch (qualityOfServiceLevel)
-            {
-                case MqttQualityOfServiceLevel.AtMostOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS0;
-                case MqttQualityOfServiceLevel.AtLeastOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS1;
-                case MqttQualityOfServiceLevel.ExactlyOnce: return MqttSubscribeReturnCode.SuccessMaximumQoS2;
-                default: return MqttSubscribeReturnCode.Failure;
-            }
-        }
     }
 }

Frameworks/MQTTnet.NetStandard/Server/MqttClientUnSubscribedTopicEventArgs.cs

@@ -0,0 +1,17 @@
+using System;
+
+namespace MQTTnet.Server
+{
+    public class MqttClientUnsubscribedTopicEventArgs : EventArgs
+    {
+        public MqttClientUnsubscribedTopicEventArgs(string clientId, string topicFilter)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
+        }
+
+        public string ClientId { get; }
+
+        public string TopicFilter { get; }
+    }
+}

Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs

@@ -8,7 +8,7 @@ using MQTTnet.Packets;
 
 namespace MQTTnet.Server
 {
-    public sealed class MqttRetainedMessagesManager
+    public sealed class MqttRetainedMessagesManager : IDisposable
     {
         private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
         private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
@@ -102,6 +102,11 @@ namespace MQTTnet.Server
             return retainedMessages;
         }
 
+        public void Dispose()
+        {
+            _semaphore?.Dispose();
+        }
+
         private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
         {
             var saveIsRequired = false;

Frameworks/MQTTnet.NetStandard/Server/MqttServer.cs

@@ -20,19 +20,19 @@ namespace MQTTnet.Server
 
         public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
         {
+            if (adapters == null) throw new ArgumentNullException(nameof(adapters));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
-            if (adapters == null)
-            {
-                throw new ArgumentNullException(nameof(adapters));
-            }
-
             _adapters = adapters.ToList();
         }
 
         public event EventHandler<MqttServerStartedEventArgs> Started;
+
         public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
         public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
+        public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
+        public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
+
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
         public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
@@ -40,6 +40,22 @@ namespace MQTTnet.Server
             return _clientSessionsManager.GetConnectedClientsAsync();
         }
 
+        public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            return _clientSessionsManager.SubscribeAsync(clientId, topicFilters);
+        }
+
+        public Task UnsubscribeAsync(string clientId, IList<string> topicFilters)
+        {
+            if (clientId == null) throw new ArgumentNullException(nameof(clientId));
+            if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
+
+            return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
+        }
+
         public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
         {
             if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
@@ -59,11 +75,19 @@ namespace MQTTnet.Server
             if (_cancellationTokenSource != null) throw new InvalidOperationException("The server is already started.");
 
             _cancellationTokenSource = new CancellationTokenSource();
-            _retainedMessagesManager = new MqttRetainedMessagesManager(_options, _logger);
-            _clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, this, _logger);
 
+            _retainedMessagesManager = new MqttRetainedMessagesManager(_options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync();
 
+            _clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, _logger)
+            {
+                ClientConnectedCallback = OnClientConnected,
+                ClientDisconnectedCallback = OnClientDisconnected,
+                ClientSubscribedTopicCallback = OnClientSubscribedTopic,
+                ClientUnsubscribedTopicCallback = OnClientUnsubscribedTopic,
+                ApplicationMessageReceivedCallback = OnApplicationMessageReceived
+            };
+
             foreach (var adapter in _adapters)
             {
                 adapter.ClientAccepted += OnClientAccepted;
@@ -98,39 +122,46 @@ namespace MQTTnet.Server
             }
             finally
             {
+                _clientSessionsManager?.Dispose();
+                _retainedMessagesManager?.Dispose();
+
                 _cancellationTokenSource = null;
                 _retainedMessagesManager = null;
                 _clientSessionsManager = null;
             }
         }
 
-        internal void OnClientConnected(ConnectedMqttClient client)
+        private void OnClientConnected(ConnectedMqttClient client)
         {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-
             _logger.Info<MqttServer>("Client '{0}': Connected.", client.ClientId);
             ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(client));
         }
 
-        internal void OnClientDisconnected(ConnectedMqttClient client)
+        private void OnClientDisconnected(ConnectedMqttClient client)
         {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-
             _logger.Info<MqttServer>("Client '{0}': Disconnected.", client.ClientId);
             ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(client));
         }
-
+        
-        internal void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
+        private void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
         {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+            ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
+        }
 
+        private void OnClientUnsubscribedTopic(string clientId, string topicFilter)
+        {
+            ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
+        }
+        
+        private void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
+        {
             ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));
         }
 
         private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
             eventArgs.SessionTask = Task.Run(
-                async () => await _clientSessionsManager.RunClientSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
+                async () => await _clientSessionsManager.RunSessionAsync(eventArgs.Client, _cancellationTokenSource.Token).ConfigureAwait(false),
                 _cancellationTokenSource.Token);
         }
     }

Frameworks/MQTTnet.Netstandard/MQTTnet.NetStandard.csproj

@@ -0,0 +1,65 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">netstandard1.3;netstandard2.0;net452;net461;uap10.0</TargetFrameworks>
+    <TargetFrameworks Condition=" '$(OS)' != 'Windows_NT' ">netstandard1.3;netstandard2.0</TargetFrameworks>
+    <AssemblyName>MQTTnet</AssemblyName>
+    <RootNamespace>MQTTnet</RootNamespace>
+    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
+    <DebugType>Full</DebugType>
+    <AssemblyVersion>0.0.0.0</AssemblyVersion>
+    <FileVersion>0.0.0.0</FileVersion>
+    <Version>0.0.0.0</Version>
+    <Company />
+    <Product />
+    <Description />
+    <Authors />
+    <PackageId />
+    <SignAssembly>true</SignAssembly>
+    <AssemblyOriginatorKeyFile>codeSigningKey.pfx</AssemblyOriginatorKeyFile>
+    <DelaySign>false</DelaySign>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
+    <CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
+    <NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
+    <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
+    <TargetPlatformVersion>10.0.16299.0</TargetPlatformVersion>
+    <TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
+    <TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
+    <TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
+    <DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
+    <LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x86'" />
+
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" />
+
+  <PropertyGroup Condition="'$(Configuration)|$(TargetFramework)|$(Platform)'=='Release|netstandard1.3|AnyCPU'">
+    <DefineConstants>RELEASE;NETSTANDARD1_3</DefineConstants>
+  </PropertyGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
+    <PackageReference Include="System.Net.Security" Version="4.3.2" />
+    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
+    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)'=='netstandard1.3'">
+    <PackageReference Include="System.Net.Security" Version="4.3.2" />
+    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
+    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
+    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="5.4.1" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)'=='net452'">
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)'=='net461'">
+  </ItemGroup>
+
+</Project>

MQTTnet.sln.DotSettings

@@ -1,7 +1,13 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
+	<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSORHOLDER_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
 	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_CALL_PARENTHESES/@EntryValue">False</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_METHOD_PARENTHESES/@EntryValue">False</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_WITHIN_PARENTHESES/@EntryValue">False</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpRenamePlacementToArrangementMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EAddAccessorOwnerDeclarationBracesMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

README.md

@@ -18,12 +18,12 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
 * TLS 1.2 support for client and server (but not UWP servers)
 * Extensible communication channels (i.e. In-Memory, TCP, TCP+TLS, WS)
 * Lightweight (only the low level implementation of MQTT, no overhead)
-* Performance optimized (processing ~30.000 messages / second)*
+* Performance optimized (processing ~40.000 messages / second)*
 * Interfaces included for mocking and testing
 * Access to internal trace messages
 * Unit tested (70+ tests)
 
-\* Tested on local machine with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
+\* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
 
 ### Client
 

Tests/MQTTnet.Core.Tests/MqttClientTests.cs

@@ -0,0 +1,36 @@
+using System.Net.Sockets;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.Core.Tests
+{
+    [TestClass]
+    public class MqttClientTests
+    {
+
+        [TestMethod]
+        public async Task ClientDisconnectException()
+        {
+            var factory = new MqttFactory();
+            var client = factory.CreateMqttClient();
+
+            var exceptionIsCorrect = false;
+            client.Disconnected += (s, e) =>
+            {
+                exceptionIsCorrect = e.Exception is MqttCommunicationException c && c.InnerException is SocketException;
+            };
+
+            try
+            {
+                await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("wrong-server").Build());
+            }
+            catch
+            {
+            }
+            
+            Assert.IsTrue(exceptionIsCorrect);
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttPacketIdentifierProviderTests.cs

@@ -0,0 +1,32 @@
+using Microsoft.VisualStudio.TestTools .UnitTesting;
+using MQTTnet.Client;
+
+namespace MQTTnet.Core.Tests
+{
+    [TestClass]
+    public class MqttPacketIdentifierProviderTests
+    {
+        [TestMethod]
+        public void Reset()
+        {
+            var p = new MqttPacketIdentifierProvider();
+            Assert.AreEqual(1, p.GetNewPacketIdentifier());
+            Assert.AreEqual(2, p.GetNewPacketIdentifier());
+            p.Reset();
+            Assert.AreEqual(1, p.GetNewPacketIdentifier());
+        }
+
+        [TestMethod]
+        public void ReachBoundaries()
+        {
+            var p = new MqttPacketIdentifierProvider();
+
+            for (ushort i = 0; i < ushort.MaxValue; i++)
+            {
+                Assert.AreEqual(i + 1, p.GetNewPacketIdentifier());
+            }
+
+            Assert.AreEqual(1, p.GetNewPacketIdentifier());
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs

@@ -12,7 +12,7 @@ namespace MQTTnet.Core.Tests
         public void MqttPacketReader_EmptyStream()
         {
             var memStream = new MemoryStream();
-            var header = MqttPacketReader.ReadHeaderFromSource(memStream, CancellationToken.None);
+            var header = MqttPacketReader.ReadHeaderAsync(memStream, CancellationToken.None).GetAwaiter().GetResult();
 
             Assert.IsNull(header);
         }

Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs

@@ -405,11 +405,11 @@ namespace MQTTnet.Core.Tests
 
             using (var headerStream = new MemoryStream(Join(buffer1)))
             {
-                var header = MqttPacketReader.ReadHeaderFromSource(headerStream, CancellationToken.None);
+                var header = MqttPacketReader.ReadHeaderAsync(headerStream, CancellationToken.None).GetAwaiter().GetResult();
 
                 using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.BodyLength))
                 {
-                    var deserializedPacket = serializer.Deserialize(header, bodyStream.ToArray());
+                    var deserializedPacket = serializer.Deserialize(header, bodyStream);
                     var buffer2 = serializer.Serialize(deserializedPacket);
 
                     Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -79,7 +79,7 @@ namespace MQTTnet.Core.Tests
         }
 
         [TestMethod]
-        public async Task MqttServer_Unsubscribe()
+        public async Task MqttServer_SubscribeUnsubscribe()
         {
             var serverAdapter = new TestMqttServerAdapter();
             var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
@@ -99,13 +99,31 @@ namespace MQTTnet.Core.Tests
                 await c2.PublishAsync(message);
                 Assert.AreEqual(0, receivedMessagesCount);
 
+                var subscribeEventCalled = false;
+                s.ClientSubscribedTopic += (_, e) =>
+                    {
+                        subscribeEventCalled = e.TopicFilter.Topic == "a" && e.ClientId == "c1";
+                    };
+
                 await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce));
+                await Task.Delay(100);
+                Assert.IsTrue(subscribeEventCalled, "Subscribe event not called.");
+
                 await c2.PublishAsync(message);
 
                 await Task.Delay(500);
                 Assert.AreEqual(1, receivedMessagesCount);
 
+                var unsubscribeEventCalled = false;
+                s.ClientUnsubscribedTopic += (_, e) =>
+                {
+                    unsubscribeEventCalled = e.TopicFilter == "a" && e.ClientId == "c1";
+                };
+
                 await c1.UnsubscribeAsync("a");
+                await Task.Delay(100);
+                Assert.IsTrue(unsubscribeEventCalled, "Unsubscribe event not called.");
+
                 await c2.PublishAsync(message);
 
                 await Task.Delay(500);
@@ -299,7 +317,7 @@ namespace MQTTnet.Core.Tests
 
             var serverAdapter = new TestMqttServerAdapter();
             var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
-            
+
             try
             {
                 var options = new MqttServerOptions { ApplicationMessageInterceptor = Interceptor };
@@ -330,6 +348,41 @@ namespace MQTTnet.Core.Tests
             }
         }
 
+        [TestMethod]
+        public async Task MqttServer_Body()
+        {
+            var serverAdapter = new TestMqttServerAdapter();
+            var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
+
+            var bodyIsMatching = false;
+            try
+            {
+                await s.StartAsync(new MqttServerOptions());
+
+                var c1 = await serverAdapter.ConnectTestClient(s, "c1");
+                var c2 = await serverAdapter.ConnectTestClient(s, "c2");
+
+                c1.ApplicationMessageReceived += (_, e) =>
+                {
+                    if (Encoding.UTF8.GetString(e.ApplicationMessage.Payload) == "The body")
+                    {
+                        bodyIsMatching = true;
+                    }
+                };
+
+                await c1.SubscribeAsync("A", MqttQualityOfServiceLevel.AtMostOnce);
+                await c2.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("A").WithPayload(Encoding.UTF8.GetBytes("The body")).Build());
+
+                await Task.Delay(500);
+            }
+            finally
+            {
+                await s.StopAsync();
+            }
+
+            Assert.IsTrue(bodyIsMatching);
+        }
+
         private class TestStorage : IMqttServerStorage
         {
             private IList<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();

Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs

@@ -11,12 +11,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeSingleSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -32,12 +32,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeDifferentQoSSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce));
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -53,13 +53,13 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeTwoTimesSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce));
             sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtLeastOnce));
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -75,12 +75,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeSingleNoSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {
@@ -94,12 +94,12 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle()
         {
-            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions());
+            var sm = new MqttClientSubscriptionsManager(new MqttServerOptions(), "");
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
 
-            sm.SubscribeAsync(sp, "").Wait();
+            sm.SubscribeAsync(sp).Wait();
 
             var pp = new MqttApplicationMessage
             {

Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj

@@ -10,7 +10,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.3" />
+    <PackageReference Include="Microsoft.AspNetCore.All" Version="2.0.5" />
   </ItemGroup>
 
   <ItemGroup>

Tests/MQTTnet.TestApp.AspNetCore2/package.json

@@ -6,7 +6,7 @@
   },
 
   "dependencies": {
-    "mqtt": "2.13.1",
+    "mqtt": "2.15.1",
     "@types/node": "8.0.46",
     "systemjs": "0.20.19",
     "typescript": "2.3.4"

Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj

@@ -138,10 +138,10 @@
   </ItemGroup>
   <ItemGroup>
     <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
-      <Version>5.4.0</Version>
+      <Version>5.4.1</Version>
     </PackageReference>
     <PackageReference Include="Microsoft.Toolkit.Uwp.UI.Controls">
-      <Version>2.0.0</Version>
+      <Version>2.1.1</Version>
     </PackageReference>
   </ItemGroup>
   <PropertyGroup Condition=" '$(VisualStudioVersion)' == '' or '$(VisualStudioVersion)' &lt; '14.0' ">