Fix issue with connection management.

Add support for V3.1.0; Performance improvements
Fix wrong QoS level handling for server.
2026-10-15 17:00:03 +08:00 · 2017-08-17 22:49:16 +02:00 · 2017-08-13 19:46:32 +02:00 · 2017-08-07 20:28:49 +02:00 · 2017-08-05 11:38:45 +02:00 · 2017-08-05 11:38:24 +02:00
68 changed files with 960 additions and 598 deletions
@@ -1,5 +1,7 @@
 ## Ignore Visual Studio temporary files, build results, and
 ## files generated by popular Visual Studio add-ons.
+##
+## Get latest from https://github.com/github/gitignore/blob/master/VisualStudio.gitignore

 # User-specific files
 *.suo
@@ -15,12 +17,12 @@
 [Dd]ebugPublic/
 [Rr]elease/
 [Rr]eleases/
-[Xx]64/
-[Xx]86/
-[Bb]uild/
+x64/
+x86/
 bld/
 [Bb]in/
 [Oo]bj/
+[Ll]og/

 # Visual Studio 2015 cache/options directory
 .vs/
@@ -40,9 +42,11 @@ TestResult.xml
 [Rr]eleasePS/
 dlldata.c

-# DNX
+# .NET Core
 project.lock.json
+project.fragment.lock.json
 artifacts/
+**/Properties/launchSettings.json

 *_i.c
 *_p.c
@@ -81,6 +85,7 @@ ipch/
 *.sdf
 *.cachefile
 *.VC.db
+*.VC.VC.opendb

 # Visual Studio profiler
 *.psess
@@ -108,6 +113,10 @@ _TeamCity*
 # DotCover is a Code Coverage Tool
 *.dotCover

+# Visual Studio code coverage results
+*.coverage
+*.coveragexml
+
 # NCrunch
 _NCrunch_*
 .*crunch*.local.xml
@@ -139,13 +148,16 @@ publish/
 # Publish Web Output
 *.[Pp]ublish.xml
 *.azurePubxml
-
-# TODO: Un-comment the next line if you do not want to checkin 
-# your web deploy settings because they may include unencrypted
-# passwords
-#*.pubxml
+# TODO: Comment the next line if you want to checkin your web deploy settings
+# but database connection strings (with potential passwords) will be unencrypted
+*.pubxml
 *.publishproj

+# Microsoft Azure Web App publish settings. Comment the next line if you want to
+# checkin your Azure Web App publish settings, but sensitive information contained
+# in these scripts will be unencrypted
+PublishScripts/
+
 # NuGet Packages
 *.nupkg
 # The packages folder can be ignored because of Package Restore
@@ -154,7 +166,7 @@ publish/
 !**/packages/build/
 # Uncomment if necessary however generally it will be regenerated when needed
 #!**/packages/repositories.config
-# NuGet v3's project.json files produces more ignoreable files
+# NuGet v3's project.json files produces more ignorable files
 *.nuget.props
 *.nuget.targets

@@ -166,12 +178,11 @@ csx/
 ecf/
 rcf/

-# Microsoft Azure ApplicationInsights config file
-ApplicationInsights.config
-
-# Windows Store app package directory
+# Windows Store app package directories and files
 AppPackages/
 BundleArtifacts/
+Package.StoreAssociation.xml
+_pkginfo.txt

 # Visual Studio cache files
 # files ending in .cache can be ignored
@@ -181,16 +192,19 @@ BundleArtifacts/

 # Others
 ClientBin/
-[Ss]tyle[Cc]op.*
 ~$*
 *~
 *.dbmdl
 *.dbproj.schemaview
+*.jfm
 *.pfx
 *.publishsettings
-node_modules/
 orleans.codegen.cs

+# Since there are multiple workflows, uncomment next line to ignore bower_components
+# (https://github.com/github/gitignore/pull/1529#issuecomment-104372622)
+#bower_components/
+
 # RIA/Silverlight projects
 Generated_Code/

@@ -205,6 +219,7 @@ UpgradeLog*.htm
 # SQL Server files
 *.mdf
 *.ldf
+*.ndf

 # Business Intelligence projects
 *.rdl.data
@@ -219,6 +234,10 @@ FakesAssemblies/

 # Node.js Tools for Visual Studio
 .ntvs_analysis.dat
+node_modules/
+
+# Typescript v1 declaration files
+typings/

 # Visual Studio 6 build log
 *.plg
@@ -226,6 +245,9 @@ FakesAssemblies/
 # Visual Studio 6 workspace options file
 *.opt

+# Visual Studio 6 auto-generated workspace file (contains which files were open etc.)
+*.vbw
+
 # Visual Studio LightSwitch build output
 **/*.HTMLClient/GeneratedArtifacts
 **/*.DesktopClient/GeneratedArtifacts
@@ -234,13 +256,34 @@ FakesAssemblies/
 **/*.Server/ModelManifest.xml
 _Pvt_Extensions

-# LightSwitch generated files
-GeneratedArtifacts/
-ModelManifest.xml
-
 # Paket dependency manager
 .paket/paket.exe
+paket-files/

 # FAKE - F# Make
 .fake/
-nuget.exe
+
+# JetBrains Rider
+.idea/
+*.sln.iml
+
+# CodeRush
+.cr/
+
+# Python Tools for Visual Studio (PTVS)
+__pycache__/
+*.pyc
+
+# Cake - Uncomment if you are using it
+# tools/**
+# !tools/packages.config
+
+# Telerik's JustMock configuration file
+*.jmconfig
+
+# BizTalk build output
+*.btp.cs
+*.btm.cs
+*.odx.cs
+*.xsd.cs
+Build/nuget.exe
@@ -0,0 +1,49 @@
+<?xml version="1.0"?>
+<package >
+  <metadata>
+    <id>MQTTnet</id>
+    <version>2.2.0</version>
+    <authors>Christian Kratky</authors>
+    <owners>Christian Kratky</owners>
+    <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
+    <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
+    <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>* [Server] Added support for MQTT protocol version 3.1.0
+* [Server] Providing the used protocol version of connected clients
+* [Client] Added support for protocol version 3.1.0
+* [Core] Several minor performance improvements
+* [Core] Fixed an issue with connection management (Thanks to wuzhenda; Zuendelmeister)
+    </releaseNotes>
+    <copyright>Copyright Christian Kratky 2016-2017</copyright>
+    <tags>MQTT MQTTClient MQTTServer MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Queue Hardware Arduino</tags> 
+    <dependencies>
+
+      <group targetFramework="netstandard1.3">
+        <dependency id="NETStandard.Library" version="1.6.1" />
+        <dependency id="System.Net.Security" version="4.3.1" />
+      </group>
+      
+      <group targetFramework="uap10.0">
+        <dependency id="Microsoft.NETCore.UniversalWindowsPlatform" version="5.3.3" />
+      </group>
+
+    </dependencies>
+  </metadata>
+
+  <files>
+    <!-- .NET Standard 1.3 -->
+    <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard1.3\MQTTnet.Core.*" target="lib\netstandard1.3\"/>
+    <file src="..\Frameworks\MQTTnet.Netstandard\bin\Release\netstandard1.3\MQTTnet.*" target="lib\netstandard1.3\"/>
+
+    <!-- Universal Windows -->
+    <file src="..\Frameworks\MQTTnet.UniversalWindows\bin\Release\MQTTnet.Core.*" target="lib\uap10.0\"/>
+    <file src="..\Frameworks\MQTTnet.UniversalWindows\bin\Release\MQTTnet.*" target="lib\uap10.0\"/>
+
+    <!-- .NET Framework -->
+    <file src="..\Frameworks\MQTTnet.NetFramework\bin\Release\MQTTnet.Core.*" target="lib\net45\"/>
+    <file src="..\Frameworks\MQTTnet.NetFramework\bin\Release\MQTTnet.*" target="lib\net45\"/>
+    
+  </files>
+</package>
@@ -0,0 +1,12 @@
+param([string]$version)
+
+if ([string]::IsNullOrEmpty($version)) {$version = "0.0.1"}
+
+$msbuild = "MSBuild.exe"
+&$msbuild ..\Frameworks\MQTTnet.NetFramework\MQTTnet.NetFramework.csproj /t:Build /p:Configuration="Release"
+&$msbuild ..\Frameworks\MQTTnet.Netstandard\MQTTnet.Netstandard.csproj /t:Build /p:Configuration="Release"
+&$msbuild ..\Frameworks\MQTTnet.UniversalWindows\MQTTnet.UniversalWindows.csproj /t:Build /p:Configuration="Release"
+
+Remove-Item .\NuGet -Force -Recurse
+New-Item -ItemType Directory -Force -Path .\NuGet
+.\NuGet.exe pack MQTTnet.nuspec -Verbosity detailed -Symbols -OutputDir "NuGet" -Version $version
@@ -1,34 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <TargetFramework>netcoreapp1.1</TargetFramework>
-    <Authors>Christian Kratky</Authors>
-    <Company>Christian Kratky</Company>
-    <Product>MQTTnet</Product>
-    <Description>MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</Description>
-    <Copyright>Copyright © Christian Kratky 2016-2017</Copyright>
-    <Version>2.1.1.0</Version>
-    <AssemblyVersion>2.1.1.0</AssemblyVersion>
-    <FileVersion>2.1.1.0</FileVersion>
-    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
-    <AssemblyName>MQTTnet</AssemblyName>
-    <RootNamespace>MQTTnet</RootNamespace>
-    <PackageId>MQTTnet.NetCore</PackageId>
-    <PackageProjectUrl>https://github.com/chkr1011/MQTTnet</PackageProjectUrl>
-    <RepositoryUrl>https://github.com/chkr1011/MQTTnet</RepositoryUrl>
-    <PackageTags>MQTT MQTTClient MQTTServer MQTTBroker Broker</PackageTags>
-    <PackageIconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</PackageIconUrl>
-    <PackageLicenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</PackageLicenseUrl>
-  </PropertyGroup>
-
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Any CPU|AnyCPU'" />
-
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Any CPU|x64'" />
-
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Any CPU|x86'" />
-
-  <ItemGroup>
-    <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
-  </ItemGroup>
-
-</Project>
@@ -63,6 +63,7 @@ namespace MQTTnet.Implementations
        {
            _isRunning = false;

+            _cancellationTokenSource?.Cancel(false);
            _cancellationTokenSource?.Dispose();
            _cancellationTokenSource = null;

@@ -85,10 +86,10 @@ 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 MqttPacketSerializer());
                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                }
-                catch (Exception exception)
+                catch (Exception exception) when (!(exception is ObjectDisposedException))
                {
                    MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while acceping connection at default endpoint.");
                }
@@ -106,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 MqttPacketSerializer());
                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                }
                catch (Exception exception)
@@ -12,12 +12,11 @@ namespace MQTTnet.Implementations
 {
    public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
    {
-        private readonly Socket _socket;
+        private Socket _socket;
        private SslStream _sslStream;

        public MqttTcpChannel()
        {
-            _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
        }

        public MqttTcpChannel(Socket socket, SslStream sslStream)
@@ -31,6 +30,11 @@ namespace MQTTnet.Implementations
            if (options == null) throw new ArgumentNullException(nameof(options));
            try
            {
+                if (_socket == null)
+                {
+                    _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+                }
+
                await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, options.Server, options.GetPort(), null);

                if (options.TlsOptions.UseTls)
@@ -49,8 +53,7 @@ namespace MQTTnet.Implementations
        {
            try
            {
-                _sslStream.Dispose();
-                _socket.Dispose();
+                Dispose();
                return Task.FromResult(0);
            }
            catch (SocketException exception)
@@ -108,6 +111,9 @@ namespace MQTTnet.Implementations
        {
            _socket?.Dispose();
            _sslStream?.Dispose();
+
+            _socket = null;
+            _sslStream = null;
        }

        private static X509CertificateCollection LoadCertificates(MqttClientOptions options)
@@ -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 MqttPacketSerializer()));
        }
    }
 }
@@ -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));

@@ -11,5 +11,5 @@ using System.Runtime.InteropServices;
 [assembly: AssemblyCulture("")]
 [assembly: ComVisible(false)]
 [assembly: Guid("a480ef90-0eaa-4d9a-b271-47a9c47f6f7d")]
-[assembly: AssemblyVersion("2.1.1.0")]
-[assembly: AssemblyFileVersion("2.1.1.0")]
+[assembly: AssemblyVersion("2.1.5.1")]
+[assembly: AssemblyFileVersion("2.1.5.1")]
@@ -61,6 +61,7 @@ namespace MQTTnet.Implementations
        {
            _isRunning = false;

+            _cancellationTokenSource?.Cancel(false);
            _cancellationTokenSource?.Dispose();
            _cancellationTokenSource = null;

@@ -83,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 MqttPacketSerializer());
                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                }
                catch (Exception exception)
@@ -104,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 MqttPacketSerializer());
                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
                }
                catch (Exception exception)
@@ -12,12 +12,11 @@ namespace MQTTnet.Implementations
 {
    public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
    {
-        private readonly Socket _socket;
+        private Socket _socket;
        private SslStream _sslStream;

        public MqttTcpChannel()
        {
-            _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
        }

        public MqttTcpChannel(Socket socket, SslStream sslStream)
@@ -31,6 +30,11 @@ namespace MQTTnet.Implementations
            if (options == null) throw new ArgumentNullException(nameof(options));
            try
            {
+                if (_socket == null)
+                {
+                    _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+                }
+
                await _socket.ConnectAsync(options.Server, options.GetPort());
                
                if (options.TlsOptions.UseTls)
@@ -49,8 +53,7 @@ namespace MQTTnet.Implementations
        {
            try
            {
-                _sslStream.Dispose();
-                _socket.Dispose();
+                Dispose();
                return Task.FromResult(0);
            }
            catch (SocketException exception)
@@ -101,6 +104,9 @@ namespace MQTTnet.Implementations
        {
            _socket?.Dispose();
            _sslStream?.Dispose();
+
+            _socket = null;
+            _sslStream = null;
        }

        private static X509CertificateCollection LoadCertificates(MqttClientOptions options)
@@ -0,0 +1,29 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netstandard1.3</TargetFramework>
+    <AssemblyName>MQTTnet</AssemblyName>
+    <RootNamespace>MQTTnet</RootNamespace>
+    <AssemblyVersion>2.1.5.1</AssemblyVersion>
+    <FileVersion>2.1.5.1</FileVersion>
+    <Version>0.0.0.0</Version>
+    <Company />
+    <Product />
+    <Description />
+    <Authors />
+    <PackageId />
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x86'" />
+
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'" />
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="System.Net.Security" Version="4.3.1" />
+  </ItemGroup>
+
+</Project>
@@ -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 MqttPacketSerializer()));
        }
    }
 }
@@ -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));

@@ -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 MqttPacketSerializer());
                ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(args.Socket.Information.RemoteAddress.ToString(), clientAdapter));
            }
            catch (Exception exception)
@@ -15,11 +15,10 @@ namespace MQTTnet.Implementations
 {
    public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
    {
-        private readonly StreamSocket _socket;
+        private StreamSocket _socket;

        public MqttTcpChannel()
        {
-            _socket = new StreamSocket();
        }

        public MqttTcpChannel(StreamSocket socket)
@@ -32,6 +31,11 @@ namespace MQTTnet.Implementations
            if (options == null) throw new ArgumentNullException(nameof(options));
            try
            {
+                if (_socket == null)
+                {
+                    _socket = new StreamSocket();
+                }
+
                if (!options.TlsOptions.UseTls)
                {
                    await _socket.ConnectAsync(new HostName(options.Server), options.GetPort().ToString());
@@ -59,7 +63,7 @@ namespace MQTTnet.Implementations
        {
            try
            {
-                _socket.Dispose();
+                Dispose();
                return Task.FromResult(0);
            }
            catch (SocketException exception)
@@ -100,6 +104,8 @@ namespace MQTTnet.Implementations
        public void Dispose()
        {
            _socket?.Dispose();
+
+            _socket = null;
        }

        private static Certificate LoadCertificate(MqttClientOptions options)
@@ -11,11 +11,12 @@
    <AssemblyName>MQTTnet</AssemblyName>
    <DefaultLanguage>en-US</DefaultLanguage>
    <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
-    <TargetPlatformVersion>10.0.14393.0</TargetPlatformVersion>
+    <TargetPlatformVersion>10.0.15063.0</TargetPlatformVersion>
    <TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
    <MinimumVisualStudioVersion>14</MinimumVisualStudioVersion>
    <FileAlignment>512</FileAlignment>
    <ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
+    <RuntimeIdentifiers>win10-arm;win10-arm-aot;win10-x86;win10-x86-aot;win10-x64;win10-x64-aot</RuntimeIdentifiers>
  </PropertyGroup>
  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
    <PlatformTarget>AnyCPU</PlatformTarget>
@@ -108,10 +109,6 @@
    <GenerateLibraryLayout>false</GenerateLibraryLayout>
    <DocumentationFile>bin\x64\Release\MQTTnet.XML</DocumentationFile>
  </PropertyGroup>
-  <ItemGroup>
-    <!-- A reference to the entire .Net Framework and Windows SDK are automatically included -->
-    <None Include="project.json" />
-  </ItemGroup>
  <ItemGroup>
    <Compile Include="MqttClientFactory.cs" />
    <Compile Include="Implementations\MqttServerAdapter.cs" />
@@ -126,6 +123,11 @@
      <Name>MQTTnet.Core</Name>
    </ProjectReference>
  </ItemGroup>
+  <ItemGroup>
+    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
+      <Version>5.3.3</Version>
+    </PackageReference>
+  </ItemGroup>
  <PropertyGroup Condition=" '$(VisualStudioVersion)' == '' or '$(VisualStudioVersion)' &lt; '14.0' ">
    <VisualStudioVersion>14.0</VisualStudioVersion>
  </PropertyGroup>
@@ -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 MqttPacketSerializer()));
        }
    }
 }
@@ -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));

@@ -10,5 +10,5 @@ using System.Runtime.InteropServices;
 [assembly: AssemblyTrademark("")]
 [assembly: AssemblyCulture("")]
 [assembly: ComVisible(false)]
-[assembly: AssemblyVersion("2.1.1.0")]
-[assembly: AssemblyFileVersion("2.1.1.0")]
+[assembly: AssemblyVersion("2.1.5.1")]
+[assembly: AssemblyFileVersion("2.1.5.1")]
@@ -1,17 +0,0 @@
-{
-  "dependencies": {
-    "Microsoft.NETCore.UniversalWindowsPlatform": "5.1.0"
-  },
-  "frameworks": {
-    "uap10.0": {}
-  },
-  "runtimes": {
-    "win10": {},
-    "win10-arm": {},
-    "win10-arm-aot": {},
-    "win10-x86": {},
-    "win10-x86-aot": {},
-    "win10-x64": {},
-    "win10-x64-aot": {}
-  }
-}
@@ -2,6 +2,7 @@
 using System.Threading.Tasks;
 using MQTTnet.Core.Client;
 using MQTTnet.Core.Packets;
+using MQTTnet.Core.Serializer;

 namespace MQTTnet.Core.Adapter
 {
@@ -14,5 +15,7 @@ namespace MQTTnet.Core.Adapter
        Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout);

        Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout);
+
+        IMqttPacketSerializer PacketSerializer { get; }
    }
 }
@@ -11,22 +11,19 @@ namespace MQTTnet.Core.Adapter
 {
    public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
    {
-        private readonly IMqttPacketSerializer _serializer;
        private readonly IMqttCommunicationChannel _channel;

        public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer)
        {
            _channel = channel ?? throw new ArgumentNullException(nameof(channel));
-            _serializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
+            PacketSerializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
        }

+        public IMqttPacketSerializer PacketSerializer { get; }
+
        public async Task ConnectAsync(MqttClientOptions options, TimeSpan timeout)
        {
-            var task = _channel.ConnectAsync(options);
-            if (await Task.WhenAny(Task.Delay(timeout), task) != task)
-            {
-                throw new MqttCommunicationTimedOutException();
-            }
+            await ExecuteWithTimeoutAsync(_channel.ConnectAsync(options), timeout);
        }

        public async Task DisconnectAsync()
@@ -38,21 +35,7 @@ namespace MQTTnet.Core.Adapter
        {
            MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"TX >>> {packet} [Timeout={timeout}]");

-            bool hasTimeout;
-            try
-            {
-                var task = _serializer.SerializeAsync(packet, _channel);
-                hasTimeout = await Task.WhenAny(Task.Delay(timeout), task) != task;
-            }
-            catch (Exception exception)
-            {
-                throw new MqttCommunicationException(exception);
-            }
-
-            if (hasTimeout)
-            {
-                throw new MqttCommunicationTimedOutException();
-            }
+            await ExecuteWithTimeoutAsync(PacketSerializer.SerializeAsync(packet, _channel), timeout);
        }

        public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout)
@@ -60,20 +43,11 @@ namespace MQTTnet.Core.Adapter
            MqttBasePacket packet;
            if (timeout > TimeSpan.Zero)
            {
-                var workerTask = _serializer.DeserializeAsync(_channel);
-                var timeoutTask = Task.Delay(timeout);
-                var hasTimeout = Task.WhenAny(timeoutTask, workerTask) == timeoutTask;
-
-                if (hasTimeout)
-                {
-                    throw new MqttCommunicationTimedOutException();
-                }
-
-                packet = workerTask.Result;
+                packet = await ExecuteWithTimeoutAsync(PacketSerializer.DeserializeAsync(_channel), timeout);
            }
            else
            {
-                packet = await _serializer.DeserializeAsync(_channel);
+                packet = await PacketSerializer.DeserializeAsync(_channel);
            }

            if (packet == null)
@@ -84,5 +58,35 @@ namespace MQTTnet.Core.Adapter
            MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"RX <<< {packet}");
            return packet;
        }
+
+        private static async Task<TResult> ExecuteWithTimeoutAsync<TResult>(Task<TResult> task, TimeSpan timeout)
+        {
+            var timeoutTask = Task.Delay(timeout);
+            if (await Task.WhenAny(timeoutTask, task) == timeoutTask)
+            {
+                throw new MqttCommunicationTimedOutException();
+            }
+
+            if (task.IsFaulted)
+            {
+                throw new MqttCommunicationException(task.Exception);
+            }
+
+            return task.Result;
+        }
+
+        private static async Task ExecuteWithTimeoutAsync(Task task, TimeSpan timeout)
+        {
+            var timeoutTask = Task.Delay(timeout);
+            if (await Task.WhenAny(timeoutTask, task) == timeoutTask)
+            {
+                throw new MqttCommunicationTimedOutException();
+            }
+
+            if (task.IsFaulted)
+            {
+                throw new MqttCommunicationException(task.Exception);
+            }
+        }
    }
 }
@@ -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);
+    }
+}
@@ -1,6 +1,6 @@
 using System;

-namespace MQTTnet.Core
+namespace MQTTnet.Core.Client
 {
    public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
    {
@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -13,15 +12,15 @@ 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;
        private readonly IMqttCommunicationAdapter _adapter;

+        private bool _disconnectedEventSuspended;
        private int _latestPacketIdentifier;
        private CancellationTokenSource _cancellationTokenSource;

@@ -29,6 +28,8 @@ namespace MQTTnet.Core.Client
        {
            _options = options ?? throw new ArgumentNullException(nameof(options));
            _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter));
+
+            _adapter.PacketSerializer.ProtocolVersion = options.ProtocolVersion;
        }

        public event EventHandler Connected;
@@ -48,50 +49,64 @@ namespace MQTTnet.Core.Client
                throw new MqttProtocolViolationException("It is not allowed to connect with a server after the connection is established.");
            }

-            var connectPacket = new MqttConnectPacket
+            try
            {
-                ClientId = _options.ClientId,
-                Username = _options.UserName,
-                Password = _options.Password,
-                CleanSession = _options.CleanSession,
-                KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds,
-                WillMessage = willApplicationMessage
-            };
+                _disconnectedEventSuspended = false;

-            await _adapter.ConnectAsync(_options, _options.DefaultCommunicationTimeout);
-            MqttTrace.Verbose(nameof(MqttClient), "Connection with server established.");
+                await _adapter.ConnectAsync(_options, _options.DefaultCommunicationTimeout);

-            _cancellationTokenSource = new CancellationTokenSource();
-            _latestPacketIdentifier = 0;
-            _processedPublishPackets.Clear();
-            _packetDispatcher.Reset();
-            IsConnected = true;
+                MqttTrace.Verbose(nameof(MqttClient), "Connection with server established.");

-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-            Task.Run(() => ReceivePackets(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+                var connectPacket = new MqttConnectPacket
+                {
+                    ClientId = _options.ClientId,
+                    Username = _options.UserName,
+                    Password = _options.Password,
+                    CleanSession = _options.CleanSession,
+                    KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds,
+                    WillMessage = willApplicationMessage
+                };

-            var response = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket);
-            if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
-            {
-                await DisconnectAsync();
-                throw new MqttConnectingFailedException(response.ConnectReturnCode);
+                _cancellationTokenSource = new CancellationTokenSource();
+                _latestPacketIdentifier = 0;
+                _packetDispatcher.Reset();
+
+                StartReceivePackets();
+
+                var response = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket);
+                if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
+                {
+                    await DisconnectInternalAsync();
+                    throw new MqttConnectingFailedException(response.ConnectReturnCode);
+                }
+
+                if (_options.KeepAlivePeriod != TimeSpan.Zero)
+                {
+                    StartSendKeepAliveMessages();
+                }
+
+                MqttTrace.Verbose(nameof(MqttClient), "MQTT connection with server established.");
+
+                IsConnected = true;
+                Connected?.Invoke(this, EventArgs.Empty);
            }
-
-            if (_options.KeepAlivePeriod != TimeSpan.Zero)
+            catch (Exception)
            {
-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-                Task.Run(() => SendKeepAliveMessagesAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+                await DisconnectInternalAsync();
+                throw;
            }
-
-            Connected?.Invoke(this, EventArgs.Empty);
        }

        public async Task DisconnectAsync()
        {
-            await SendAsync(new MqttDisconnectPacket());
-            await DisconnectInternalAsync();
+            try
+            {
+                await SendAsync(new MqttDisconnectPacket());
+            }
+            finally
+            {
+                await DisconnectInternalAsync();
+            }
        }

        public Task<IList<MqttSubscribeResult>> SubscribeAsync(params TopicFilter[] topicFilters)
@@ -105,6 +120,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 +170,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 +181,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>());
            }
        }

@@ -180,14 +197,23 @@ namespace MQTTnet.Core.Client
            {
                await _adapter.DisconnectAsync();
            }
-            catch
+            catch (Exception exception)
            {
+                MqttTrace.Warning(nameof(MqttClient), exception, "Error while disconnecting.");
            }
            finally
            {
-                _cancellationTokenSource?.Cancel();
+                _cancellationTokenSource?.Cancel(false);
+                _cancellationTokenSource?.Dispose();
+                _cancellationTokenSource = null;
+
                IsConnected = false;
-                Disconnected?.Invoke(this, EventArgs.Empty);
+
+                if (!_disconnectedEventSuspended)
+                {
+                    _disconnectedEventSuspended = true;
+                    Disconnected?.Invoke(this, EventArgs.Empty);
+                }
            }
        }

@@ -205,14 +231,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);
                }
@@ -229,19 +253,16 @@ namespace MQTTnet.Core.Client

        private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
        {
-            if (publishPacket.QualityOfServiceLevel != MqttQualityOfServiceLevel.AtMostOnce)
+            var applicationMessage = publishPacket.ToApplicationMessage();
+
+            try
            {
-                _processedPublishPackets.Add(publishPacket.PacketIdentifier);
+                ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while handling application message.");
            }
-
-            var applicationMessage = new MqttApplicationMessage(
-                publishPacket.Topic,
-                publishPacket.Payload,
-                publishPacket.QualityOfServiceLevel,
-                publishPacket.Retain
-            );
-
-            ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage));
        }

        private Task ProcessReceivedPublishPacket(MqttPublishPacket publishPacket)
@@ -260,24 +281,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)
@@ -295,18 +319,15 @@ 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)
+                if (pi1 == null || pi2 == null)
                {
-                    if (pi1.PacketIdentifier != pi2.PacketIdentifier)
-                    {
-                        return false;
-                    }
+                    return true;
                }

-                return true;
+                return pi1.PacketIdentifier == pi2.PacketIdentifier;
            }

            await _adapter.SendPacketAsync(requestPacket, _options.DefaultCommunicationTimeout);
@@ -333,15 +354,16 @@ namespace MQTTnet.Core.Client
            catch (MqttCommunicationException exception)
            {
                MqttTrace.Warning(nameof(MqttClient), exception, "MQTT communication error while receiving packets.");
+                await DisconnectInternalAsync();
            }
            catch (Exception exception)
            {
                MqttTrace.Warning(nameof(MqttClient), exception, "Error while sending/receiving keep alive packets.");
+                await DisconnectInternalAsync();
            }
            finally
            {
                MqttTrace.Information(nameof(MqttClient), "Stopped sending keep alive packets.");
-                await DisconnectInternalAsync();
            }
        }

@@ -352,27 +374,47 @@ namespace MQTTnet.Core.Client
            {
                while (!cancellationToken.IsCancellationRequested)
                {
-                    var mqttPacket = await _adapter.ReceivePacketAsync(TimeSpan.Zero);
-                    MqttTrace.Information(nameof(MqttClient), $"Received <<< {mqttPacket}");
+                    var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero);
+                    MqttTrace.Information(nameof(MqttClient), $"Received <<< {packet}");

-#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
-                    Task.Run(() => ProcessReceivedPacketAsync(mqttPacket), cancellationToken);
-#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+                    StartProcessReceivedPacket(packet, cancellationToken);
                }
            }
            catch (MqttCommunicationException exception)
            {
-                MqttTrace.Warning(nameof(MqttClient), exception, "MQTT communication error while receiving packets.");
+                MqttTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while receiving packets.");
+                await DisconnectInternalAsync();
            }
            catch (Exception exception)
            {
-                MqttTrace.Error(nameof(MqttClient), exception, "Error while receiving packets.");
+                MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while receiving packets.");
+                await DisconnectInternalAsync();
            }
            finally
            {
                MqttTrace.Information(nameof(MqttClient), "Stopped receiving packets.");
-                await DisconnectInternalAsync();
            }
        }
+
+        private void StartProcessReceivedPacket(MqttBasePacket packet, CancellationToken cancellationToken)
+        {
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            Task.Run(() => ProcessReceivedPacketAsync(packet), cancellationToken);
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+        }
+
+        private void StartReceivePackets()
+        {
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            Task.Run(() => ReceivePackets(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+        }
+
+        private void StartSendKeepAliveMessages()
+        {
+#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+            Task.Run(() => SendKeepAliveMessagesAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
+#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
+        }
    }
 }
@@ -1,4 +1,5 @@
 using System;
+using MQTTnet.Core.Serializer;

 namespace MQTTnet.Core.Client
 {
@@ -7,8 +8,8 @@ namespace MQTTnet.Core.Client
        public string Server { get; set; }

        public int? Port { get; set; }
-        
-        public MqttClientTlsOptions TlsOptions { get; } = new MqttClientTlsOptions();
+
+        public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();

        public string UserName { get; set; }

@@ -21,5 +22,7 @@ namespace MQTTnet.Core.Client
        public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5);

        public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
+
+        public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
    }
 }
@@ -4,7 +4,7 @@ namespace MQTTnet.Core.Exceptions
 {
    public class MqttCommunicationException : Exception
    {
-        public MqttCommunicationException()
+        protected MqttCommunicationException()
        {
        }

@@ -17,5 +17,10 @@ namespace MQTTnet.Core.Exceptions
            : base(message)
        {
        }
+
+        public MqttCommunicationException(string message, Exception innerException)
+            : base(message, innerException)
+        {
+        }
    }
 }
@@ -4,6 +4,16 @@ namespace MQTTnet.Core.Internal
 {
    internal static class MqttApplicationMessageExtensions
    {
+        public static MqttApplicationMessage ToApplicationMessage(this MqttPublishPacket publishPacket)
+        {
+            return new MqttApplicationMessage(
+                publishPacket.Topic,
+                publishPacket.Payload,
+                publishPacket.QualityOfServiceLevel,
+                publishPacket.Retain
+            );
+        }
+
        public static MqttPublishPacket ToPublishPacket(this MqttApplicationMessage applicationMessage)
        {
            if (applicationMessage == null)
@@ -5,20 +5,20 @@
    <AssemblyName>MQTTnet.Core</AssemblyName>
    <RootNamespace>MQTTnet.Core</RootNamespace>
    <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
-    <Description>MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</Description>
-    <Product>MQTTnet</Product>
-    <Company>Christian Kratky</Company>
-    <Authors>Christian Kratky</Authors>
-    <Version>2.1.2.0</Version>
-    <PackageId>MQTTnet.Core</PackageId>
-    <Copyright>Copyright © Christian Kratky 2016-2017</Copyright>
-    <PackageProjectUrl>https://github.com/chkr1011/MQTTnet</PackageProjectUrl>
-    <PackageIconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</PackageIconUrl>
-    <RepositoryUrl>https://github.com/chkr1011/MQTTnet</RepositoryUrl>
-    <PackageTags>MQTT MQTTClient MQTTServer MQTTBroker Broker</PackageTags>
-    <FileVersion>2.1.2.0</FileVersion>
-    <AssemblyVersion>2.1.2.0</AssemblyVersion>
-    <PackageLicenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</PackageLicenseUrl>
+    <Description></Description>
+    <Product></Product>
+    <Company></Company>
+    <Authors></Authors>
+    <Version>0.0.0.0</Version>
+    <PackageId></PackageId>
+    <Copyright></Copyright>
+    <PackageProjectUrl></PackageProjectUrl>
+    <PackageIconUrl></PackageIconUrl>
+    <RepositoryUrl></RepositoryUrl>
+    <PackageTags></PackageTags>
+    <FileVersion>2.1.5.1</FileVersion>
+    <AssemblyVersion>2.1.5.1</AssemblyVersion>
+    <PackageLicenseUrl></PackageLicenseUrl>
  </PropertyGroup>

  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Any CPU|AnyCPU'" />
@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Packets
 {
-    public interface IPacketWithIdentifier
+    public interface IMqttPacketWithIdentifier
    {
        ushort PacketIdentifier { get; set; }
    }
@@ -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;
-        }
    }
 }
@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Packets
 {
-    public class MqttBasePublishPacket : MqttBasePacket, IPacketWithIdentifier
+    public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier
    {
        public ushort PacketIdentifier { get; set; }
    }
@@ -1,7 +1,11 @@
-namespace MQTTnet.Core.Packets
+using MQTTnet.Core.Serializer;
+
+namespace MQTTnet.Core.Packets
 {
    public sealed class MqttConnectPacket: MqttBasePacket
    {
+        public MqttProtocolVersion ProtocolVersion { get; set; }
+
        public string ClientId { get; set; }

        public string Username { get; set; }
@@ -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;
+        }
+    }
+}
@@ -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; }

@@ -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; }
        
@@ -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; }
    }
@@ -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; }
        
@@ -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;
        }
    }
 }
@@ -6,6 +6,8 @@ namespace MQTTnet.Core.Serializer
 {
    public interface IMqttPacketSerializer
    {
+        MqttProtocolVersion ProtocolVersion { get; set; }
+
        Task SerializeAsync(MqttBasePacket mqttPacket, IMqttCommunicationChannel destination);

        Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source);
@@ -10,7 +10,7 @@ namespace MQTTnet.Core.Serializer
 {
    public sealed class MqttPacketReader : IDisposable
    {
-        private readonly MemoryStream _remainingData = new MemoryStream();
+        private readonly MemoryStream _remainingData = new MemoryStream(1024);
        private readonly IMqttCommunicationChannel _source;

        private int _remainingLength;
@@ -9,93 +9,84 @@ using MQTTnet.Core.Protocol;

 namespace MQTTnet.Core.Serializer
 {
-    public sealed class DefaultMqttV311PacketSerializer : IMqttPacketSerializer
+    public sealed class MqttPacketSerializer : IMqttPacketSerializer
    {
+        private static byte[] ProtocolVersionV311Name { get; } = Encoding.UTF8.GetBytes("MQTT");
+        private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs");
+
+        public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
+
        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 +197,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 +212,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 +229,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,18 +257,24 @@ 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

+            MqttProtocolVersion protocolVersion;
            var protocolName = await reader.ReadRemainingDataAsync(4);
-
-            if (Encoding.UTF8.GetString(protocolName, 0, protocolName.Length) != "MQTT")
+            if (protocolName.SequenceEqual(ProtocolVersionV310Name))
            {
-                throw new MqttProtocolViolationException("Protocol name is not 'MQTT'.");
+                await reader.ReadRemainingDataAsync(2);
+                protocolVersion = MqttProtocolVersion.V310;
+            }
+            else if (protocolName.SequenceEqual(ProtocolVersionV311Name))
+            {
+                protocolVersion = MqttProtocolVersion.V311;
+            }
+            else
+            {
+                throw new MqttProtocolViolationException("Protocol name is not supported.");
            }

            var protocolLevel = await reader.ReadRemainingDataByteAsync();
@@ -285,7 +282,13 @@ namespace MQTTnet.Core.Serializer

            var connectFlagsReader = new ByteReader(connectFlags);
            connectFlagsReader.Read(); // Reserved.
-            packet.CleanSession = connectFlagsReader.Read();
+
+            var packet = new MqttConnectPacket
+            {
+                ProtocolVersion = protocolVersion,
+                CleanSession = connectFlagsReader.Read()
+            };
+
            var willFlag = connectFlagsReader.Read();
            var willQoS = connectFlagsReader.Read(2);
            var willRetain = connectFlagsReader.Read();
@@ -318,7 +321,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 +336,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 +350,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 +358,7 @@ namespace MQTTnet.Core.Serializer
            }
        }

-        private void ValidatePublishPacket(MqttPublishPacket packet)
+        private static void ValidatePublishPacket(MqttPublishPacket packet)
        {
            if (packet.QualityOfServiceLevel == 0 && packet.Dup)
            {
@@ -363,8 +366,6 @@ namespace MQTTnet.Core.Serializer
            }
        }

-        private static readonly byte[] MqttPrefix = Encoding.UTF8.GetBytes("MQTT");
-
        private Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
        {
            ValidateConnectPacket(packet);
@@ -373,9 +374,19 @@ namespace MQTTnet.Core.Serializer
            {
                // Write variable header
                output.Write(0x00, 0x04); // 3.1.2.1 Protocol Name
-                output.Write(MqttPrefix);
-                output.Write(0x04); // 3.1.2.2 Protocol Level
-
+                if (ProtocolVersion == MqttProtocolVersion.V311)
+                {
+                    output.Write(ProtocolVersionV311Name);
+                    output.Write(0x04); // 3.1.2.2 Protocol Level (4)
+                }
+                else
+                {
+                    output.Write(ProtocolVersionV310Name);
+                    output.Write(0x64);
+                    output.Write(0x70);
+                    output.Write(0x03); // Protocol Level (3)
+                }
+                
                var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags
                connectFlags.Write(false); // Reserved
                connectFlags.Write(packet.CleanSession);
@@ -425,8 +436,12 @@ namespace MQTTnet.Core.Serializer
            using (var output = new MqttPacketWriter())
            {
                var connectAcknowledgeFlags = new ByteWriter();
-                connectAcknowledgeFlags.Write(packet.IsSessionPresent);

+                if (ProtocolVersion == MqttProtocolVersion.V311)
+                {
+                    connectAcknowledgeFlags.Write(packet.IsSessionPresent);
+                }
+                
                output.Write(connectAcknowledgeFlags);
                output.Write((byte)packet.ConnectReturnCode);

@@ -435,22 +450,33 @@ namespace MQTTnet.Core.Serializer
            }
        }

-        private Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination)
+        private static async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination)
+        {
+            using (var output = new MqttPacketWriter())
+            {
+                output.Write(packet.PacketIdentifier);
+
+                output.InjectFixedHeader(MqttControlPacketType.PubRel, 0x02);
+                await output.WriteToAsync(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 +511,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 +522,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,18 +533,7 @@ namespace MQTTnet.Core.Serializer
            }
        }

-        private async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
-            {
-                output.Write(packet.PacketIdentifier);
-
-                output.InjectFixedHeader(MqttControlPacketType.PubRel, 0x02);
-                await output.WriteToAsync(destination);
-            }
-        }
-
-        private Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
+        private static Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
        {
            using (var output = new MqttPacketWriter())
            {
@@ -529,7 +544,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 +564,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 +583,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 +602,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 +613,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())
            {
@@ -9,46 +9,13 @@ namespace MQTTnet.Core.Serializer
 {
    public sealed class MqttPacketWriter : IDisposable
    {
-        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);
-        }
+        private readonly MemoryStream _buffer = new MemoryStream(1024);

        public void InjectFixedHeader(MqttControlPacketType packetType, byte flags = 0)
        {
-            var fixedHeader = (byte)((byte)packetType << 4);
+            var fixedHeader = (int)packetType << 4;
            fixedHeader |= flags;
-            InjectFixedHeader(fixedHeader);
+            InjectFixedHeader((byte)fixedHeader);
        }

        public void Write(byte value)
@@ -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);
+        }
    }
 }
@@ -0,0 +1,8 @@
+namespace MQTTnet.Core.Serializer
+{
+    public enum MqttProtocolVersion
+    {
+        V311,
+        V310
+    }
+}
@@ -0,0 +1,11 @@
+using MQTTnet.Core.Serializer;
+
+namespace MQTTnet.Core.Server
+{
+    public class ConnectedMqttClient
+    {
+        public string ClientId { get; set; }
+
+        public MqttProtocolVersion ProtocolVersion { get; set; }
+    }
+}
@@ -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<ConnectedMqttClient> GetConnectedClients();
+        void InjectClient(string identifier, IMqttCommunicationAdapter adapter);
+        void Publish(MqttApplicationMessage applicationMessage);
+        void Start();
+        void Stop();
+    }
+}
@@ -0,0 +1,17 @@
+using System;
+
+namespace MQTTnet.Core.Server
+{
+    public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
+    {
+        public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
+        {
+            ClientId = clientId;
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+        }
+
+        public string ClientId { get; }
+
+        public MqttApplicationMessage ApplicationMessage { get; }
+    }
+}
@@ -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();
            }
        }
@@ -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; }
@@ -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,7 +13,7 @@ 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;
@@ -21,19 +21,23 @@ namespace MQTTnet.Core.Server
        private readonly MqttServerOptions _options;

        private CancellationTokenSource _cancellationTokenSource;
-        private IMqttCommunicationAdapter _adapter;
        private string _identifier;
        private MqttApplicationMessage _willApplicationMessage;

-        public MqttClientSession(MqttServerOptions options, Action<MqttClientSession, MqttPublishPacket> publishPacketReceivedCallback)
+        public MqttClientSession(string clientId, MqttServerOptions options, Action<MqttClientSession, MqttPublishPacket> publishPacketReceivedCallback)
        {
+            ClientId = clientId;
            _options = options ?? throw new ArgumentNullException(nameof(options));
            _publishPacketReceivedCallback = publishPacketReceivedCallback ?? throw new ArgumentNullException(nameof(publishPacketReceivedCallback));

            _messageQueue = new MqttClientMessageQueue(options);
        }

-        public bool IsConnected => _adapter != null;
+        public string ClientId { get; }
+
+        public bool IsConnected => Adapter != null;
+
+        public IMqttCommunicationAdapter Adapter { get; private set; }

        public async Task RunAsync(string identifier, MqttApplicationMessage willApplicationMessage, IMqttCommunicationAdapter adapter)
        {
@@ -44,7 +48,7 @@ namespace MQTTnet.Core.Server
            try
            {
                _identifier = identifier;
-                _adapter = adapter;
+                Adapter = adapter;
                _cancellationTokenSource = new CancellationTokenSource();

                _messageQueue.Start(adapter);
@@ -70,23 +74,22 @@ namespace MQTTnet.Core.Server

                _messageQueue.Stop();
                _cancellationTokenSource.Cancel();
-                _adapter = null;
+                Adapter = null;

                MqttTrace.Information(nameof(MqttClientSession), $"Client '{_identifier}': Disconnected.");
            }
        }

-        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.");
        }

@@ -98,39 +101,40 @@ 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);
+                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);
+                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)
            {
-                return _adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
+                return Adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
            }

            if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
@@ -145,39 +149,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);
        }
    }
 }
@@ -5,6 +5,7 @@ using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Diagnostics;
 using MQTTnet.Core.Exceptions;
+using MQTTnet.Core.Internal;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;

@@ -21,6 +22,8 @@ namespace MQTTnet.Core.Server
            _options = options ?? throw new ArgumentNullException(nameof(options));
        }

+        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+
        public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
        {
            try
@@ -31,6 +34,9 @@ namespace MQTTnet.Core.Server
                    throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
                }

+                // Switch to the required protocol version before sending any response.
+                eventArgs.ClientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion;
+
                var connectReturnCode = ValidateConnection(connectPacket);
                if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
                {
@@ -70,11 +76,15 @@ namespace MQTTnet.Core.Server
            }
        }

-        public IList<string> GetConnectedClients()
+        public IList<ConnectedMqttClient> GetConnectedClients()
        {
            lock (_syncRoot)
            {
-                return _clientSessions.Where(s => s.Value.IsConnected).Select(s => s.Key).ToList();
+                return _clientSessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
+                {
+                    ClientId = s.Value.ClientId,
+                    ProtocolVersion = s.Value.Adapter.PacketSerializer.ProtocolVersion
+                }).ToList();
            }
        }

@@ -114,7 +124,7 @@ namespace MQTTnet.Core.Server
                {
                    isExistingSession = false;

-                    clientSession = new MqttClientSession(_options, DispatchPublishPacket);
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, DispatchPublishPacket);
                    _clientSessions[connectPacket.ClientId] = clientSession;

                    MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Created a new session for client '{connectPacket.ClientId}'.");
@@ -124,11 +134,24 @@ namespace MQTTnet.Core.Server
            }
        }

-        private void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
        {
-            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);
+                }
            }
        }
    }
@@ -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;
@@ -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;
@@ -21,15 +22,25 @@ namespace MQTTnet.Core.Server
            _adapters = adapters ?? throw new ArgumentNullException(nameof(adapters));
            
            _clientSessionsManager = new MqttClientSessionsManager(options);
+            _clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
        }

-        public IList<string> GetConnectedClients()
+        public IList<ConnectedMqttClient> GetConnectedClients()
        {
            return _clientSessionsManager.GetConnectedClients();
        }

        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;

+        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));
@@ -56,7 +67,8 @@ namespace MQTTnet.Core.Server

        public void Stop()
        {
-            _cancellationTokenSource?.Cancel();
+            _cancellationTokenSource?.Cancel(false);
+            _cancellationTokenSource?.Dispose();
            _cancellationTokenSource = null;

            foreach (var adapter in _adapters)
@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Server
 {
-    public sealed class DefaultEndpointOptions
+    public sealed class MqttServerDefaultEndpointOptions
    {
        public bool IsEnabled { get; set; } = true;

@@ -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; }
    }
@@ -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;
-        }
-    }
-}
@@ -1,56 +0,0 @@
-<?xml version="1.0"?>
-<package >
-  <metadata>
-    <id>MQTTnet</id>
-    <version>2.1.3.0</version>
-    <authors>Christian Kratky</authors>
-    <owners>Christian Kratky</owners>
-    <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
-    <projectUrl>https://github.com/chkr1011/MQTTnet</projectUrl>
-    <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>* [Core] Any exception while accessing the underlying data source is now rethrown as "MqttCommunicationException".
-* [Core] Extended exception information when an invalid packet type is received.
-* [Server] Added TLS 1.2 support (but not for UWP) (thanks to Zazzmatazz)
-* [Client] Added TLS 1.2 support (thanks to Zazzmatazz)
-* [Core] Optimized async task management
-* [Nuget] Fixed x64 files</releaseNotes>
-    <copyright>Copyright Christian Kratky 2016-2017</copyright>
-    <tags>MQTT MQTTClient MQTTServer MQTTBroker Broker</tags>
-  </metadata>
-
-  <files>
-    <!-- Universal Windows -->
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\Release\MQTTnet.Core.dll" target="lib\uap10.0"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\Release\MQTTnet.dll" target="lib\uap10.0"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\Release\MQTTnet.pri" target="lib\uap10.0"/>
-
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\ARM\Release\MQTTnet.Core.dll" target="lib\uap10.0-arm"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\ARM\Release\MQTTnet.dll" target="lib\uap10.0-arm"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\ARM\Release\MQTTnet.pri" target="lib\uap10.0-arm"/>
-
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\x64\Release\MQTTnet.Core.dll" target="lib\uap10.0-x64"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\x64\Release\MQTTnet.dll" target="lib\uap10.0-x64"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\x64\Release\MQTTnet.pri" target="lib\uap10.0-x64"/>
-
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\x86\Release\MQTTnet.Core.dll" target="lib\uap10.0-x86"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\x86\Release\MQTTnet.dll" target="lib\uap10.0-x86"/>
-    <file src=".\Frameworks\MQTTnet.UniversalWindows\bin\x86\Release\MQTTnet.pri" target="lib\uap10.0-x86"/>
-
-    <!-- .NET Framework -->
-    <file src=".\Frameworks\MQTTnet.NetFramework\bin\Release\MQTTnet.Core.dll" target="lib\net45"/>
-    <file src=".\Frameworks\MQTTnet.NetFramework\bin\Release\MQTTnet.dll" target="lib\net45"/>
-    
-    <file src=".\Frameworks\MQTTnet.NetFramework\bin\x86\Release\MQTTnet.Core.dll" target="lib\net45-x86"/>
-    <file src=".\Frameworks\MQTTnet.NetFramework\bin\x86\Release\MQTTnet.dll" target="lib\net45-x86"/>
-
-    <file src=".\Frameworks\MQTTnet.NetFramework\bin\x64\Release\MQTTnet.Core.dll" target="lib\net45-x64"/>
-    <file src=".\Frameworks\MQTTnet.NetFramework\bin\x64\Release\MQTTnet.dll" target="lib\net45-x64"/>
-
-    <!-- .NET Core App -->
-    <file src=".\Frameworks\MQTTnet.NetCoreApp\bin\Release\netcoreapp1.1\MQTTnet.Core.dll" target="lib\netcoreapp1.1"/>
-    <file src=".\Frameworks\MQTTnet.NetCoreApp\bin\Release\netcoreapp1.1\MQTTnet.dll" target="lib\netcoreapp1.1"/>
-    
-  </files>
-</package>
@@ -1,7 +1,7 @@

 Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio 15
-VisualStudioVersion = 15.0.26430.12
+VisualStudioVersion = 15.0.26430.16
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Core.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Core.Tests.csproj", "{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}"
 EndProject
@@ -13,19 +13,25 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Tests", "Tests", "{9248C2E1
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Frameworks", "Frameworks", "{32A630A7-2598-41D7-B625-204CD906F5FB}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Core", "MQTTnet.Core\MQTTnet.Core.csproj", "{2ECB99E4-72D0-4C23-99BA-93D511D3967D}"
-EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{002203AF-2565-4C0D-95ED-027FDEFE0C35}"
-	ProjectSection(SolutionItems) = preProject
-		MQTTnet.nuspec = MQTTnet.nuspec
-	EndProjectSection
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.NetCoreApp", "Frameworks\MQTTnet.NetCoreApp\MQTTnet.NetCoreApp.csproj", "{1A1B7F51-5328-4395-9D9C-07D70965825E}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Core", "MQTTnet.Core\MQTTnet.Core.csproj", "{2ECB99E4-72D0-4C23-99BA-93D511D3967D}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.TestApp.NetFramework", "Tests\MQTTnet.TestApp.NetFramework\MQTTnet.TestApp.NetFramework.csproj", "{D9D74F33-6943-49B2-B765-7BD589082098}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.TestApp.UniversalWindows", "Tests\MQTTnet.TestApp.UniversalWindows\MQTTnet.TestApp.UniversalWindows.csproj", "{FF1F72D6-9524-4422-9497-3CC0002216ED}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.NetStandard", "Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj", "{3587E506-55A2-4EB3-99C7-DC01E42D25D2}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{67C28AC1-BC3A-420A-BE9C-FA2401431CF9}"
+	ProjectSection(SolutionItems) = preProject
+		Build\build.ps1 = Build\build.ps1
+		Build\MQTTnet.nuspec = Build\MQTTnet.nuspec
+	EndProjectSection
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}"
+	ProjectSection(SolutionItems) = preProject
+		README.md = README.md
+	EndProjectSection
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -71,6 +77,7 @@ Global
 		{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Release|x86.ActiveCfg = Release|x86
 		{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Release|x86.Build.0 = Release|x86
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|ARM.ActiveCfg = Debug|ARM
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|ARM.Build.0 = Debug|ARM
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|x64.ActiveCfg = Debug|x64
@@ -78,6 +85,7 @@ Global
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|x86.ActiveCfg = Debug|x86
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Debug|x86.Build.0 = Debug|x86
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|Any CPU.Build.0 = Release|Any CPU
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|ARM.ActiveCfg = Release|ARM
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|ARM.Build.0 = Release|ARM
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611}.Release|x64.ActiveCfg = Release|x64
@@ -100,22 +108,6 @@ Global
 		{2ECB99E4-72D0-4C23-99BA-93D511D3967D}.Release|x64.Build.0 = Release|x64
 		{2ECB99E4-72D0-4C23-99BA-93D511D3967D}.Release|x86.ActiveCfg = Release|x86
 		{2ECB99E4-72D0-4C23-99BA-93D511D3967D}.Release|x86.Build.0 = Release|x86
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|ARM.ActiveCfg = Debug|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|ARM.Build.0 = Debug|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|x64.ActiveCfg = Debug|x64
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|x64.Build.0 = Debug|x64
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|x86.ActiveCfg = Debug|x86
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Debug|x86.Build.0 = Debug|x86
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|Any CPU.Build.0 = Release|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|ARM.ActiveCfg = Release|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|ARM.Build.0 = Release|Any CPU
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|x64.ActiveCfg = Release|x64
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|x64.Build.0 = Release|x64
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|x86.ActiveCfg = Release|x86
-		{1A1B7F51-5328-4395-9D9C-07D70965825E}.Release|x86.Build.0 = Release|x86
 		{D9D74F33-6943-49B2-B765-7BD589082098}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{D9D74F33-6943-49B2-B765-7BD589082098}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{D9D74F33-6943-49B2-B765-7BD589082098}.Debug|ARM.ActiveCfg = Debug|Any CPU
@@ -133,6 +125,7 @@ Global
 		{D9D74F33-6943-49B2-B765-7BD589082098}.Release|x86.ActiveCfg = Release|x86
 		{D9D74F33-6943-49B2-B765-7BD589082098}.Release|x86.Build.0 = Release|x86
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|ARM.ActiveCfg = Debug|ARM
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|ARM.Build.0 = Debug|ARM
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|ARM.Deploy.0 = Debug|ARM
@@ -143,6 +136,7 @@ Global
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|x86.Build.0 = Debug|x86
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Debug|x86.Deploy.0 = Debug|x86
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|Any CPU.Build.0 = Release|Any CPU
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|ARM.ActiveCfg = Release|ARM
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|ARM.Build.0 = Release|ARM
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|ARM.Deploy.0 = Release|ARM
@@ -152,6 +146,22 @@ Global
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|x86.ActiveCfg = Release|x86
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|x86.Build.0 = Release|x86
 		{FF1F72D6-9524-4422-9497-3CC0002216ED}.Release|x86.Deploy.0 = Release|x86
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|ARM.ActiveCfg = Debug|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|ARM.Build.0 = Debug|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x64.ActiveCfg = Debug|x64
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x64.Build.0 = Debug|x64
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x86.ActiveCfg = Debug|x86
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x86.Build.0 = Debug|x86
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|Any CPU.Build.0 = Release|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|ARM.ActiveCfg = Release|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|ARM.Build.0 = Release|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x64.ActiveCfg = Release|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x64.Build.0 = Release|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x86.ActiveCfg = Release|Any CPU
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -160,8 +170,8 @@ Global
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4}
 		{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D} = {32A630A7-2598-41D7-B625-204CD906F5FB}
 		{BD60C727-D8E8-40C3-B8E3-C95A864AE611} = {32A630A7-2598-41D7-B625-204CD906F5FB}
-		{1A1B7F51-5328-4395-9D9C-07D70965825E} = {32A630A7-2598-41D7-B625-204CD906F5FB}
 		{D9D74F33-6943-49B2-B765-7BD589082098} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4}
 		{FF1F72D6-9524-4422-9497-3CC0002216ED} = {9248C2E1-B9D6-40BF-81EC-86004D7765B4}
+		{3587E506-55A2-4EB3-99C7-DC01E42D25D2} = {32A630A7-2598-41D7-B625-204CD906F5FB}
 	EndGlobalSection
 EndGlobal
@@ -2,50 +2,61 @@
 <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/.
+MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). The implementation is based on the documentation from http://mqtt.org/.

-## Features
-* MQTT client included
-* MQTT server (broker) included
-* TLS 1.2 support for client and server (but not UWP servers)
+# Features
+
+## General
 * Async support
-* List of connected clients available (server only)
+* TLS 1.2 support for client and server (but not UWP servers)
 * Extensible communication channels (i.e. In-Memory, TCP, TCP+SSL, WebSockets (not included in this project))
-* Access to internal trace messages
-* Extensible client credential validation (server only)
-* Unit tested (48+ tests)
+* Interfaces included for mocking and testing
 * Lightweight (only the low level implementation of MQTT, no overhead)
+* Access to internal trace messages
+* Unit tested (50+ tests)

-## Supported frameworks
-* .NET Standard 1.1+
+## Client
+* Rx support (via another project)
+
+## Server (broker)
+* List of connected clients available
+* Supports connected clients with different protocol versions at the same time
+* Able to publish its own messages (no loopback client required)
+* Able to receive every messages (no loopback client required)
+* Extensible client credential validation
+
+# Supported frameworks
+* .NET Standard 1.3+
 * .NET Core 1.1+
 * .NET Core App 1.1+
 * .NET Framework 4.5.2+ (x86, x64, AnyCPU)
 * Universal Windows (UWP) 10.0.10240+ (x86, x64, ARM, AnyCPU)

-## Supported MQTT versions
+# Supported MQTT versions
 * 3.1.1
+* 3.1.0

-## Nuget
+# Nuget
 This library is available as a nuget package: https://www.nuget.org/packages/MQTTnet/

-## Contributions
+# Contributions
 If you want to contribute to this project just create a pull request.

-## References
+# 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.

-# MqttClient
-## Example
+# Examples
+## MqttClient

-```c#
+```csharp
 var options = new MqttClientOptions
 {
    Server = "localhost"
@@ -115,11 +126,9 @@ while (true)
 }
 ```

-# MqttServer
+## MqttServer

-## Example 
-
-```c#
+```csharp
 var options = new MqttServerOptions
 {
    ConnectionValidator = p =>
@@ -86,7 +86,7 @@
  <ItemGroup>
    <Compile Include="ByteReaderTests.cs" />
    <Compile Include="ByteWriterTests.cs" />
-    <Compile Include="DefaultMqttV311PacketSerializerTests.cs" />
+    <Compile Include="MqttPacketSerializerTests.cs" />
    <Compile Include="MqttServerTests.cs" />
    <Compile Include="MqttSubscriptionsManagerTests.cs" />
    <Compile Include="Properties\AssemblyInfo.cs" />
@@ -12,8 +12,23 @@ using MQTTnet.Core.Serializer;
 namespace MQTTnet.Core.Tests
 {
    [TestClass]
-    public class DefaultMqttV311PacketSerializerTests
+    public class MqttPacketSerializerTests
    {
+        [TestMethod]
+        public void SerializeV310_MqttConnectPacket()
+        {
+            var p = new MqttConnectPacket
+            {
+                ClientId = "XYZ",
+                Password = "PASS",
+                Username = "USER",
+                KeepAlivePeriod = 123,
+                CleanSession = true
+            };
+
+            SerializeAndCompare(p, "EB0ABE1RSXNkcAPCAHsAA1hZWgAEVVNFUgAEUEFTUw==", MqttProtocolVersion.V310);
+        }
+
        [TestMethod]
        public void SerializeV311_MqttConnectPacket()
        {
@@ -96,6 +111,17 @@ namespace MQTTnet.Core.Tests
            SerializeAndCompare(p, "IAIBBQ==");
        }

+        [TestMethod]
+        public void SerializeV310_MqttConnAckPacket()
+        {
+            var p = new MqttConnAckPacket
+            {
+                ConnectReturnCode = MqttConnectReturnCode.ConnectionAccepted
+            };
+
+            SerializeAndCompare(p, "IAIAAA==", MqttProtocolVersion.V310);
+        }
+
        [TestMethod]
        public void DeserializeV311_MqttConnAckPacket()
        {
@@ -403,9 +429,9 @@ namespace MQTTnet.Core.Tests
            }
        }

-        private void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
+        private void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
        {
-            var serializer = new DefaultMqttV311PacketSerializer();
+            var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion };
            var channel = new TestChannel();
            serializer.SerializeAsync(packet, channel).Wait();
            var buffer = channel.ToArray();
@@ -415,7 +441,7 @@ namespace MQTTnet.Core.Tests

        private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
        {
-            var serializer = new DefaultMqttV311PacketSerializer();
+            var serializer = new MqttPacketSerializer();

            var channel1 = new TestChannel();
            serializer.SerializeAsync(packet, channel1).Wait();
@@ -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();
@@ -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));
        }
    }
 }
@@ -4,6 +4,7 @@ using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Client;
 using MQTTnet.Core.Packets;
+using MQTTnet.Core.Serializer;

 namespace MQTTnet.Core.Tests
 {
@@ -13,6 +14,8 @@ namespace MQTTnet.Core.Tests

        public TestMqttCommunicationAdapter Partner { get; set; }

+        public IMqttPacketSerializer PacketSerializer { get; } = new MqttPacketSerializer();
+
        public async Task ConnectAsync(MqttClientOptions options, TimeSpan timeout)
        {
            await Task.FromResult(0);
@@ -8,6 +8,11 @@ namespace MQTTnet.Core.Tests
    {
        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;

+        public void FireClientConnectedEvent(MqttClientConnectedEventArgs eventArgs)
+        {
+            ClientConnected?.Invoke(this, eventArgs);
+        }
+
        public void Start(MqttServerOptions options)
        {
        }
@@ -11,8 +11,8 @@
    <AssemblyName>MQTTnet.TestApp.UniversalWindows</AssemblyName>
    <DefaultLanguage>en-US</DefaultLanguage>
    <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
-    <TargetPlatformVersion Condition=" '$(TargetPlatformVersion)' == '' ">10.0.14393.0</TargetPlatformVersion>
-    <TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
+    <TargetPlatformVersion Condition=" '$(TargetPlatformVersion)' == '' ">10.0.15063.0</TargetPlatformVersion>
+    <TargetPlatformMinVersion>10.0.15063.0</TargetPlatformMinVersion>
    <MinimumVisualStudioVersion>14</MinimumVisualStudioVersion>
    <FileAlignment>512</FileAlignment>
    <ProjectTypeGuids>{A5A43C5B-DE2A-4C0C-9213-0A381AF9435A};{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids>
@@ -137,7 +137,7 @@
  </ItemGroup>
  <ItemGroup>
    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
-      <Version>5.2.3</Version>
+      <Version>5.3.3</Version>
    </PackageReference>
  </ItemGroup>
  <PropertyGroup Condition=" '$(VisualStudioVersion)' == '' or '$(VisualStudioVersion)' &lt; '14.0' ">
@@ -150,7 +150,7 @@
    <NoWarn>;2008</NoWarn>
    <NoStdLib>true</NoStdLib>
    <DebugType>full</DebugType>
-    <PlatformTarget>AnyCPU</PlatformTarget>
+    <PlatformTarget>x86</PlatformTarget>
    <UseVSHostingProcess>false</UseVSHostingProcess>
    <ErrorReport>prompt</ErrorReport>
    <CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
@@ -8,7 +8,7 @@ namespace MQTTnet.TestApp.UniversalWindows
 {
    public sealed partial class MainPage
    {
-        private MqttClient _mqttClient;
+        private IMqttClient _mqttClient;

        public MainPage()
        {
Author	SHA1	Message	Date
Christian Kratky	c90bdbfb93	Fix issue with connection management.	2017-08-17 22:49:16 +02:00
Christian Kratky	598ed66a3b	Add support for V3.1.0; Performance improvements	2017-08-13 19:46:32 +02:00
Christian Kratky	59b0b79a32	Fix wrong QoS level handling for server.	2017-08-07 20:28:49 +02:00
Christian Kratky	0172606747	Merge branch 'develop'	2017-08-05 11:38:45 +02:00
Christian Kratky	65c2a0fd7c	Update documentation	2017-08-05 11:38:24 +02:00
Christian Kratky	68926cef20	Add unit tests	2017-08-05 11:38:03 +02:00
Christian Kratky	447ca6cabb	Add interfaces, optimize performance, fix QoS level 2	2017-08-05 11:37:52 +02:00
Christian Kratky	0b0239bcf1	Performance refactoring.	2017-07-28 22:21:48 +02:00
Christian Kratky	e021d5a56b	Handle exceptions whicha re thrown while processing application messages.	2017-07-28 22:18:39 +02:00
Christian Kratky	36abf8450e	Merge branch 'develop'	2017-07-17 20:19:22 +02:00
Christian Kratky	d67b4a0b35	Update documentation	2017-07-17 20:19:08 +02:00
Christian Kratky	e4a376acdd	Merge branch 'master' into develop	2017-07-15 19:00:29 +02:00
Christian Kratky	8cfdb2fbf2	Update release notes	2017-07-15 18:37:22 +02:00
Christian Kratky	886b2d6b42	Fix issue when stopping server	2017-07-15 18:36:33 +02:00
Christian Kratky	2193281c6a	Add version information for next release	2017-07-15 18:36:09 +02:00
Christian Kratky	eb7ee420fc	Regorganize files	2017-07-15 10:16:26 +02:00
Christian Kratky	5a1d953278	Move build files to build folder	2017-07-15 10:11:03 +02:00
Christian	67c291e15a	Merge pull request #15 from 1iveowl/master Merge fixes	2017-07-15 09:59:48 +02:00
1iveowl	3722bc9ae6	Folder fix	2017-07-14 08:26:25 +02:00
1iveowl	1fb0944e2b	Folder fix	2017-07-14 08:26:03 +02:00
1iveowl	61c7ee0958	Merge branch 'master' of https://github.com/1iveowl/MQTTnet	2017-07-14 08:23:45 +02:00
1iveowl	9c5e5b5e10	build fix	2017-07-14 08:22:41 +02:00
1iveowl	b3e11d47b7	Build fix	2017-07-14 08:22:11 +02:00
1iveowl	31419fef03	Update .gitignore	2017-07-14 08:22:02 +02:00
Christian Kratky	b2e3686f59	Fix build files	2017-07-13 23:13:56 +02:00
Christian	9b90ae0c81	Merge pull request #14 from 1iveowl/master Added missing Nuspec file	2017-07-13 23:12:35 +02:00
1iveowl	39beb4912b	Added missing Nuspec file Added missing Nuspec file	2017-07-13 20:32:58 +02:00
Christian Kratky	98d7bc980c	Add an event to receive all application messages at server level.	2017-07-13 20:28:49 +02:00
Christian	2ff449b241	Merge pull request #13 from 1iveowl/master Build	2017-07-13 19:48:44 +02:00
1iveowl	83f6ae4e54	UWP target update	2017-07-13 15:01:35 +02:00
1iveowl	d26d29e9a6	PowerShell build	2017-07-13 15:00:35 +02:00
Christian Kratky	d0e8a0c4e5	Refactor .NET Standard projects	2017-07-12 19:56:57 +02:00
Christian Kratky	35344b27c5	Fixed build settings	2017-07-10 21:19:06 +02:00
Christian	6b6b7f90c8	Merge pull request #9 from 1iveowl/master Adding support for .NET Standard	2017-07-10 21:02:31 +02:00
1iveowl	eeb3242573	Fixed Assembly name	2017-07-06 00:27:19 +02:00
1iveowl	cd5c295484	Adding .NET Standard 1.3 support	2017-07-05 21:57:37 +02:00