Fix several bugs

Fix50 eventing

Build/MQTTnet.nuspec

@@ -2,7 +2,7 @@
 <package >
   <metadata>
     <id>MQTTnet</id>
-    <version>2.3.1</version>
+    <version>2.4.0</version>
     <authors>Christian Kratky</authors>
     <owners>Christian Kratky</owners>
     <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,10 +10,17 @@
     <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] Fixed an issue when accepting a new connection (UWP only) (Thanks to haeberle)
+    <releaseNotes>* [Server] Added an event which is fired when a client has disconnected.
-[Core] Fixed a dead lock while sending messages (Thanks to 1liveowl, JanEggers)
+* [Server] Added support for retained application messages
-[Client] The client is no longer sending packets before receiving has started
+* [Server] Added support for saving and loading retained messages
-[Core] Minor changes and improvements
+* [Server] The client connection is now closed if sending of one pending application message has failed
+* [Server] Fixed handling of _Dup_ flag (Thanks to haeberle)
+* [Core] Optimized exception handling
+* [Core] Mono is now also supported (Thanks to JTrotta)
+* [Client] The options are now passed in _ConnectAsync_ (Breaking change! Read Wiki for examples)
+* [Core] Trace class renamed to _MqttNetTrace_ (Breaking change!)
+* [Client] Extended certificate validation options (Breaking change!)
+* [Client] Added static certificate validation callback (NetFramework, NetStandard) / ignorable certificate errors (UniversalWindows) to _MqttTcpChannel_
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2017</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M</tags> 

Frameworks/MQTTnet.NetFramework/Implementations/MqttCommunicationAdapterFactory.cs

@@ -0,0 +1,27 @@
+using System;
+using MQTTnet.Core.Adapter;
+using MQTTnet.Core.Client;
+using MQTTnet.Core.Serializer;
+
+namespace MQTTnet.Implementations
+{
+    public class MqttCommunicationAdapterFactory : IMqttCommunicationAdapterFactory
+    {
+        public IMqttCommunicationAdapter CreateMqttCommunicationAdapter(MqttClientOptions options)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            if (options is MqttClientTcpOptions tcpOptions)
+            {
+                return new MqttChannelCommunicationAdapter(new MqttTcpChannel(tcpOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion });
+            }
+
+            if (options is MqttClientWebSocketOptions webSocketOptions)
+            {
+                return new MqttChannelCommunicationAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion });
+            }
+
+            throw new NotSupportedException();
+        }
+    }
+}

Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs

@@ -22,9 +22,9 @@ namespace MQTTnet.Implementations
 
         private bool _isRunning;
 
-        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        public event Action<IMqttCommunicationAdapter> ClientAccepted;
 
-        public void Start(MqttServerOptions options)
+        public Task StartAsync(MqttServerOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
@@ -57,9 +57,11 @@ namespace MQTTnet.Implementations
 
                 Task.Run(() => AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
             }
+
+            return Task.FromResult(0);
         }
 
-        public void Stop()
+        public Task StopAsync()
         {
             _isRunning = false;
 
@@ -72,11 +74,13 @@ namespace MQTTnet.Implementations
 
             _tlsEndpointSocket?.Dispose();
             _tlsEndpointSocket = null;
+
+            return Task.FromResult(0);
         }
 
         public void Dispose()
         {
-            Stop();
+            StopAsync();
         }
 
         private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cancellationToken)
@@ -89,11 +93,11 @@ namespace MQTTnet.Implementations
 
                     var tcpChannel = new MqttTcpChannel(clientSocket, null);
                     var clientAdapter = new MqttChannelCommunicationAdapter(tcpChannel, new MqttPacketSerializer());
-                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
+                    ClientAccepted?.Invoke(clientAdapter);
                 }
                 catch (Exception exception) when (!(exception is ObjectDisposedException))
                 {
-                    MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at default endpoint.");
+                    MqttNetTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at default endpoint.");
 
                     //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
@@ -114,11 +118,11 @@ namespace MQTTnet.Implementations
 
                     var tcpChannel = new MqttTcpChannel(clientSocket, sslStream);
                     var clientAdapter = new MqttChannelCommunicationAdapter(tcpChannel, new MqttPacketSerializer());
-                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
+                    ClientAccepted?.Invoke(clientAdapter);
                 }
                 catch (Exception exception)
                 {
-                    MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at TLS endpoint.");
+                    MqttNetTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at TLS endpoint.");
 
                     //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);

Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs

@@ -7,11 +7,14 @@ using System.Threading.Tasks;
 using MQTTnet.Core.Channel;
 using MQTTnet.Core.Client;
 using System.IO;
+using System.Linq;
 
 namespace MQTTnet.Implementations
 {
     public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
     {
+        private readonly MqttClientTcpOptions _options;
+
         // ReSharper disable once MemberCanBePrivate.Global
         // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global
         public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user.
@@ -22,9 +25,9 @@ namespace MQTTnet.Implementations
         /// <summary>
         /// called on client sockets are created in connect
         /// </summary>
-        public MqttTcpChannel()
+        public MqttTcpChannel(MqttClientTcpOptions options)
         {
-
+            _options = options ?? throw new ArgumentNullException(nameof(options));
         }
 
         /// <summary>
@@ -35,6 +38,7 @@ namespace MQTTnet.Implementations
         {
             _socket = socket ?? throw new ArgumentNullException(nameof(socket));
             _sslStream = sslStream;
+
             CreateStreams(socket, sslStream);
         }
 
@@ -42,22 +46,21 @@ namespace MQTTnet.Implementations
         public Stream ReceiveStream { get; private set; }
         public Stream RawReceiveStream { get; private set; }
 
-        public async Task ConnectAsync(MqttClientOptions options)
+        public static Func<X509Certificate, X509Chain, SslPolicyErrors, MqttClientTcpOptions, bool> CustomCertificateValidationCallback { get; set; }
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
 
+        public async Task ConnectAsync()
+        {
             if (_socket == null)
             {
                 _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
             }
 
-            await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, options.Server, options.GetPort(), null).ConfigureAwait(false);
+            await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, _options.Server, _options.GetPort(), null).ConfigureAwait(false);
 
-            if (options.TlsOptions.UseTls)
+            if (_options.TlsOptions.UseTls)
             {
-                _sslStream = new SslStream(new NetworkStream(_socket, true));
+                _sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
-
+                await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(_options), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
-                await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation).ConfigureAwait(false);
             }
 
             CreateStreams(_socket, _sslStream);
@@ -97,6 +100,37 @@ namespace MQTTnet.Implementations
             ReceiveStream = new BufferedStream(RawReceiveStream, BufferSize);
         }
 
+        private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
+        {
+            if (CustomCertificateValidationCallback != null)
+            {
+                return CustomCertificateValidationCallback(x509Certificate, chain, sslPolicyErrors, _options);
+            }
+
+            if (sslPolicyErrors == SslPolicyErrors.None)
+            {
+                return true;
+            }
+
+            if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.RevocationStatusUnknown))
+            {
+                if (!_options.TlsOptions.IgnoreCertificateRevocationErrors)
+                {
+                    return false;
+                }
+            }
+
+            if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.PartialChain))
+            {
+                if (!_options.TlsOptions.IgnoreCertificateChainErrors)
+                {
+                    return false;
+                }
+            }
+            
+            return _options.TlsOptions.AllowUntrustedCertificates;
+        }
+
         private static X509CertificateCollection LoadCertificates(MqttClientOptions options)
         {
             var certificates = new X509CertificateCollection();

Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs

@@ -10,27 +10,29 @@ namespace MQTTnet.Implementations
 {
     public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
     {
+        private readonly MqttClientWebSocketOptions _options;
+
         private ClientWebSocket _webSocket;
 
+        public MqttWebSocketChannel(MqttClientWebSocketOptions options)
+        {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
         public Stream RawReceiveStream { get; private set; }
         public Stream SendStream => RawReceiveStream;
         public Stream ReceiveStream => RawReceiveStream;
-
+        
-        public async Task ConnectAsync(MqttClientOptions options)
+        public async Task ConnectAsync()
         {
-            var uri = options.Server;
+            var uri = _options.Uri;
             if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase))
             {
                 uri = "ws://" + uri;
             }
-
+ 
-            if (options.Port.HasValue)
-            {
-                uri += ":" + options.Port;
-            }
-            
             _webSocket = new ClientWebSocket();
-            _webSocket.Options.KeepAliveInterval = options.KeepAlivePeriod;
+            _webSocket.Options.KeepAliveInterval = _options.KeepAlivePeriod;
             await _webSocket.ConnectAsync(new Uri(uri), CancellationToken.None).ConfigureAwait(false);
             RawReceiveStream = new WebSocketStream(_webSocket);
         }

Frameworks/MQTTnet.NetFramework/MQTTnet.NetFramework.csproj

@@ -100,6 +100,7 @@
     <Reference Include="System.Core" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="Implementations\MqttCommunicationAdapterFactory.cs" />
     <Compile Include="Implementations\MqttWebSocketChannel.cs" />
     <Compile Include="Implementations\WebSocketStream.cs" />
     <Compile Include="MqttClientFactory.cs" />

Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs

@@ -1,33 +1,13 @@
-using System;
+using MQTTnet.Core.Client;
-using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Channel;
-using MQTTnet.Core.Client;
-using MQTTnet.Core.Serializer;
 using MQTTnet.Implementations;
 
 namespace MQTTnet
 {
     public class MqttClientFactory : IMqttClientFactory
     {
-        public IMqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient()
         {
-            if (options == null) throw new ArgumentNullException(nameof(options));
+            return new MqttClient(new MqttCommunicationAdapterFactory());
-
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(GetMqttCommunicationChannel(options), new MqttPacketSerializer()));
-        }
-
-        private static IMqttCommunicationChannel GetMqttCommunicationChannel(MqttClientOptions options)
-        {
-            switch (options.ConnectionType)
-            {
-                case MqttConnectionType.Tcp:
-                    return new MqttTcpChannel();
-                case MqttConnectionType.Ws:
-                    return new MqttWebSocketChannel();
-
-                default:
-                    throw new NotSupportedException();
-            }
         }
     }
 }

Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs

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

Frameworks/MQTTnet.NetStandard/Implementations/MqttCommunicationAdapterFactory.cs

@@ -0,0 +1,27 @@
+using System;
+using MQTTnet.Core.Adapter;
+using MQTTnet.Core.Client;
+using MQTTnet.Core.Serializer;
+
+namespace MQTTnet.Implementations
+{
+    public class MqttCommunicationAdapterFactory : IMqttCommunicationAdapterFactory
+    {
+        public IMqttCommunicationAdapter CreateMqttCommunicationAdapter(MqttClientOptions options)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            if (options is MqttClientTcpOptions tcpOptions)
+            {
+                return new MqttChannelCommunicationAdapter(new MqttTcpChannel(tcpOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion });
+            }
+
+            if (options is MqttClientWebSocketOptions webSocketOptions)
+            {
+                return new MqttChannelCommunicationAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion });
+            }
+
+            throw new NotSupportedException();
+        }
+    }
+}

Frameworks/MQTTnet.NetStandard/Implementations/MqttServerAdapter.cs

@@ -22,9 +22,9 @@ namespace MQTTnet.Implementations
 
         private bool _isRunning;
 
-        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        public event Action<IMqttCommunicationAdapter> ClientAccepted;
 
-        public void Start(MqttServerOptions options)
+        public Task StartAsync(MqttServerOptions options)
         {
             if (_isRunning) throw new InvalidOperationException("Server is already started.");
 
@@ -56,9 +56,11 @@ namespace MQTTnet.Implementations
 
                 Task.Run(() => AcceptTlsEndpointConnectionsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
             }
+
+            return Task.FromResult(0);
         }
 
-        public void Stop()
+        public Task StopAsync()
         {
             _isRunning = false;
 
@@ -71,11 +73,13 @@ namespace MQTTnet.Implementations
 
             _tlsEndpointSocket?.Dispose();
             _tlsEndpointSocket = null;
+
+            return Task.FromResult(0);
         }
 
         public void Dispose()
         {
-            Stop();
+            StopAsync();
         }
 
         private async Task AcceptDefaultEndpointConnectionsAsync(CancellationToken cancellationToken)
@@ -86,11 +90,11 @@ namespace MQTTnet.Implementations
                 {
                     var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false);
                     var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer());
-                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
+                    ClientAccepted?.Invoke(clientAdapter);
                 }
                 catch (Exception exception)
                 {
-                    MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at default endpoint.");
+                    MqttNetTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at default endpoint.");
 
                     //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
@@ -110,11 +114,11 @@ namespace MQTTnet.Implementations
                     await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
 
                     var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer());
-                    ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
+                    ClientAccepted?.Invoke(clientAdapter);
                 }
                 catch (Exception exception)
                 {
-                    MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at TLS endpoint.");
+                    MqttNetTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at TLS endpoint.");
 
                     //excessive CPU consumed if in endless loop of socket errors
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);

Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs

@@ -7,19 +7,23 @@ using System.Threading.Tasks;
 using MQTTnet.Core.Channel;
 using MQTTnet.Core.Client;
 using System.IO;
+using System.Linq;
 
 namespace MQTTnet.Implementations
 {
     public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
     {
+        private readonly MqttClientTcpOptions _options;
+
         private Socket _socket;
         private SslStream _sslStream;
         
         /// <summary>
         /// called on client sockets are created in connect
         /// </summary>
-        public MqttTcpChannel()
+        public MqttTcpChannel(MqttClientTcpOptions options)
         {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
         }
 
         /// <summary>
@@ -36,23 +40,23 @@ namespace MQTTnet.Implementations
         public Stream SendStream => ReceiveStream;
         public Stream ReceiveStream { get; private set; }
         public Stream RawReceiveStream => ReceiveStream;
-        
-        public async Task ConnectAsync(MqttClientOptions options)
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
 
+        public static Func<X509Certificate, X509Chain, SslPolicyErrors, MqttClientTcpOptions, bool> CustomCertificateValidationCallback { get; set; }
+
+        public async Task ConnectAsync()
+        {
             if (_socket == null)
             {
                 _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
             }
 
-            await _socket.ConnectAsync(options.Server, options.GetPort()).ConfigureAwait(false);
+            await _socket.ConnectAsync(_options.Server, _options.GetPort()).ConfigureAwait(false);
 
-            if (options.TlsOptions.UseTls)
+            if (_options.TlsOptions.UseTls)
             {
-                _sslStream = new SslStream(new NetworkStream(_socket, true));
+                _sslStream = new SslStream(new NetworkStream(_socket, true), false, InternalUserCertificateValidationCallback);
                 ReceiveStream = _sslStream;
-                await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation).ConfigureAwait(false);
+                await _sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(_options), SslProtocols.Tls12, _options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
             }
             else
             {
@@ -75,6 +79,37 @@ namespace MQTTnet.Implementations
             _sslStream = null;
         }
 
+        private bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
+        {
+            if (CustomCertificateValidationCallback != null)
+            {
+                return CustomCertificateValidationCallback(x509Certificate, chain, sslPolicyErrors, _options);
+            }
+
+            if (sslPolicyErrors == SslPolicyErrors.None)
+            {
+                return true;
+            }
+
+            if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.RevocationStatusUnknown))
+            {
+                if (!_options.TlsOptions.IgnoreCertificateRevocationErrors)
+                {
+                    return false;
+                }
+            }
+
+            if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.PartialChain))
+            {
+                if (!_options.TlsOptions.IgnoreCertificateChainErrors)
+                {
+                    return false;
+                }
+            }
+
+            return _options.TlsOptions.AllowUntrustedCertificates;
+        }
+
         private static X509CertificateCollection LoadCertificates(MqttClientOptions options)
         {
             var certificates = new X509CertificateCollection();

Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs

@@ -10,27 +10,28 @@ namespace MQTTnet.Implementations
 {
     public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
     {
+        private readonly MqttClientWebSocketOptions _options;
         private ClientWebSocket _webSocket;
 
+        public MqttWebSocketChannel(MqttClientWebSocketOptions options)
+        {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
         public Stream SendStream => RawReceiveStream;
         public Stream ReceiveStream => RawReceiveStream;
         public Stream RawReceiveStream { get; private set; }
 
-        public async Task ConnectAsync(MqttClientOptions options)
+        public async Task ConnectAsync()
         {
-            var uri = options.Server;
+            var uri = _options.Uri;
             if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase))
             {
                 uri = "ws://" + uri;
             }
 
-            if (options.Port.HasValue)
-            {
-                uri += ":" + options.Port;
-            }
-            
             _webSocket = new ClientWebSocket();
-            _webSocket.Options.KeepAliveInterval = options.KeepAlivePeriod;
+            _webSocket.Options.KeepAliveInterval = _options.KeepAlivePeriod;
             await _webSocket.ConnectAsync(new Uri(uri), CancellationToken.None);
             RawReceiveStream = new WebSocketStream(_webSocket);
         }

Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj

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

Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs

@@ -1,33 +1,13 @@
-using System;
+using MQTTnet.Core.Client;
-using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Channel;
-using MQTTnet.Core.Client;
-using MQTTnet.Core.Serializer;
 using MQTTnet.Implementations;
 
 namespace MQTTnet
 {
     public class MqttClientFactory : IMqttClientFactory
     {
-        public IMqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient()
         {
-            if (options == null) throw new ArgumentNullException(nameof(options));
+            return new MqttClient(new MqttCommunicationAdapterFactory());
-
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(GetMqttCommunicationChannel(options), new MqttPacketSerializer()));
-        }
-
-        private static IMqttCommunicationChannel GetMqttCommunicationChannel(MqttClientOptions options)
-        {
-            switch (options.ConnectionType)
-            {
-                case MqttConnectionType.Tcp:
-                    return new MqttTcpChannel();
-                case MqttConnectionType.Ws:
-                    return new MqttWebSocketChannel();
-
-                default:
-                    throw new NotSupportedException();
-            }
         }
     }
 }

Frameworks/MQTTnet.UniversalWindows/Implementations/MqttCommunicationAdapterFactory.cs

@@ -0,0 +1,27 @@
+using System;
+using MQTTnet.Core.Adapter;
+using MQTTnet.Core.Client;
+using MQTTnet.Core.Serializer;
+
+namespace MQTTnet.Implementations
+{
+    public class MqttCommunicationAdapterFactory : IMqttCommunicationAdapterFactory
+    {
+        public IMqttCommunicationAdapter CreateMqttCommunicationAdapter(MqttClientOptions options)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            if (options is MqttClientTcpOptions tcpOptions)
+            {
+                return new MqttChannelCommunicationAdapter(new MqttTcpChannel(tcpOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion });
+            }
+
+            if (options is MqttClientWebSocketOptions webSocketOptions)
+            {
+                return new MqttChannelCommunicationAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion });
+            }
+
+            throw new NotSupportedException();
+        }
+    }
+}

Frameworks/MQTTnet.UniversalWindows/Implementations/MqttServerAdapter.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Diagnostics;
 using MQTTnet.Core.Serializer;
@@ -13,9 +14,9 @@ namespace MQTTnet.Implementations
 
         private bool _isRunning;
 
-        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        public event Action<IMqttCommunicationAdapter> ClientAccepted;
 
-        public void Start(MqttServerOptions options)
+        public async Task StartAsync(MqttServerOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
@@ -26,7 +27,7 @@ namespace MQTTnet.Implementations
             if (options.DefaultEndpointOptions.IsEnabled)
             {
                 _defaultEndpointSocket = new StreamSocketListener();
-                _defaultEndpointSocket.BindServiceNameAsync(options.GetDefaultEndpointPort().ToString(), SocketProtectionLevel.PlainSocket).GetAwaiter().GetResult();
+                await _defaultEndpointSocket.BindServiceNameAsync(options.GetDefaultEndpointPort().ToString(), SocketProtectionLevel.PlainSocket);
                 _defaultEndpointSocket.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync;
             }
 
@@ -36,17 +37,19 @@ namespace MQTTnet.Implementations
             }
         }
 
-        public void Stop()
+        public Task StopAsync()
         {
             _isRunning = false;
 
             _defaultEndpointSocket?.Dispose();
             _defaultEndpointSocket = null;
+
+            return Task.FromResult(0);
         }
 
         public void Dispose()
         {
-            Stop();
+            StopAsync();
         }
 
         private void AcceptDefaultEndpointConnectionsAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
@@ -54,11 +57,11 @@ namespace MQTTnet.Implementations
             try
             {
                 var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(args.Socket), new MqttPacketSerializer());
-                ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(args.Socket.Information.RemoteAddress.ToString(), clientAdapter));
+                ClientAccepted?.Invoke(clientAdapter);
             }
             catch (Exception exception)
             {
-                MqttTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at default endpoint.");
+                MqttNetTrace.Error(nameof(MqttServerAdapter), exception, "Error while accepting connection at default endpoint.");
             }
         }
     }

Frameworks/MQTTnet.UniversalWindows/Implementations/MqttTcpChannel.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.IO;
 using System.Linq;
 using System.Runtime.InteropServices.WindowsRuntime;
@@ -13,15 +14,18 @@ namespace MQTTnet.Implementations
 {
     public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
     {
+        private readonly MqttClientTcpOptions _options;
         private StreamSocket _socket;
 
-        public MqttTcpChannel()
+        public MqttTcpChannel(MqttClientTcpOptions options)
         {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
         }
 
         public MqttTcpChannel(StreamSocket socket)
         {
             _socket = socket ?? throw new ArgumentNullException(nameof(socket));
+
             CreateStreams();
         }
 
@@ -29,30 +33,29 @@ namespace MQTTnet.Implementations
         public Stream ReceiveStream { get; private set; }
         public Stream RawReceiveStream { get; private set; }
 
-        public async Task ConnectAsync(MqttClientOptions options)
+        public static Func<MqttClientTcpOptions, IEnumerable<ChainValidationResult>> CustomIgnorableServerCertificateErrorsResolver { get; set; }
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
 
+        public async Task ConnectAsync()
+        {
             if (_socket == null)
             {
                 _socket = new StreamSocket();
             }
 
-            if (!options.TlsOptions.UseTls)
+            if (!_options.TlsOptions.UseTls)
             {
-                await _socket.ConnectAsync(new HostName(options.Server), options.GetPort().ToString());
+                await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString());
             }
             else
             {
-                _socket.Control.ClientCertificate = LoadCertificate(options);
+                _socket.Control.ClientCertificate = LoadCertificate(_options);
 
-                if (!options.TlsOptions.CheckCertificateRevocation)
+                foreach (var ignorableChainValidationResult in ResolveIgnorableServerCertificateErrors())
                 {
-                    _socket.Control.IgnorableServerCertificateErrors.Add(ChainValidationResult.IncompleteChain);
+                    _socket.Control.IgnorableServerCertificateErrors.Add(ignorableChainValidationResult);
-                    _socket.Control.IgnorableServerCertificateErrors.Add(ChainValidationResult.RevocationInformationMissing);
                 }
 
-                await _socket.ConnectAsync(new HostName(options.Server), options.GetPort().ToString(), SocketProtectionLevel.Tls12);
+                await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), SocketProtectionLevel.Tls12);
             }
 
             CreateStreams();
@@ -100,5 +103,34 @@ namespace MQTTnet.Implementations
 
             return new Certificate(options.TlsOptions.Certificates.First().AsBuffer());
         }
+
+        private IEnumerable<ChainValidationResult> ResolveIgnorableServerCertificateErrors()
+        {
+            if (CustomIgnorableServerCertificateErrorsResolver != null)
+            {
+                return CustomIgnorableServerCertificateErrorsResolver(_options);
+            }
+
+            var result = new List<ChainValidationResult>();
+
+            if (_options.TlsOptions.IgnoreCertificateRevocationErrors)
+            {
+                result.Add(ChainValidationResult.RevocationInformationMissing);
+                //_socket.Control.IgnorableServerCertificateErrors.Add(ChainValidationResult.Revoked); Not supported.
+                result.Add(ChainValidationResult.RevocationFailure);
+            }
+
+            if (_options.TlsOptions.IgnoreCertificateChainErrors)
+            {
+                result.Add(ChainValidationResult.IncompleteChain);
+            }
+
+            if (_options.TlsOptions.AllowUntrustedCertificates)
+            {
+                result.Add(ChainValidationResult.Untrusted);
+            }
+
+            return result;
+        }
     }
 }

Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs

@@ -10,27 +10,28 @@ namespace MQTTnet.Implementations
 {
     public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
     {
+        private readonly MqttClientWebSocketOptions _options;
         private ClientWebSocket _webSocket = new ClientWebSocket();
 
+        public MqttWebSocketChannel(MqttClientWebSocketOptions options)
+        {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
         public Stream RawReceiveStream { get; private set; }
         public Stream SendStream => RawReceiveStream;
         public Stream ReceiveStream => RawReceiveStream;
 
-        public async Task ConnectAsync(MqttClientOptions options)
+        public async Task ConnectAsync()
         {
-            var uri = options.Server;
+            var uri = _options.Uri;
             if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase))
             {
                 uri = "ws://" + uri;
             }
 
-            if (options.Port.HasValue)
-            {
-                uri += ":" + options.Port;
-            }
-            
             _webSocket = new ClientWebSocket();
-            _webSocket.Options.KeepAliveInterval = options.KeepAlivePeriod;
+            _webSocket.Options.KeepAliveInterval = _options.KeepAlivePeriod;
             await _webSocket.ConnectAsync(new Uri(uri), CancellationToken.None).ConfigureAwait(false);
 
             RawReceiveStream = new WebSocketStream(_webSocket);

Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj

@@ -110,6 +110,7 @@
     <DocumentationFile>bin\x64\Release\MQTTnet.XML</DocumentationFile>
   </PropertyGroup>
   <ItemGroup>
+    <Compile Include="Implementations\MqttCommunicationAdapterFactory.cs" />
     <Compile Include="Implementations\MqttWebSocketChannel.cs" />
     <Compile Include="MqttClientFactory.cs" />
     <Compile Include="Implementations\MqttServerAdapter.cs" />

Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs

@@ -1,33 +1,13 @@
-using System;
+using MQTTnet.Core.Client;
-using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Channel;
-using MQTTnet.Core.Client;
-using MQTTnet.Core.Serializer;
 using MQTTnet.Implementations;
 
 namespace MQTTnet
 {
     public class MqttClientFactory : IMqttClientFactory
     {
-        public IMqttClient CreateMqttClient(MqttClientOptions options)
+        public IMqttClient CreateMqttClient()
         {
-            if (options == null) throw new ArgumentNullException(nameof(options));
+            return new MqttClient(new MqttCommunicationAdapterFactory());
-
-            return new MqttClient(options, new MqttChannelCommunicationAdapter(GetMqttCommunicationChannel(options), new MqttPacketSerializer()));
-        }
-
-        private static IMqttCommunicationChannel GetMqttCommunicationChannel(MqttClientOptions options)
-        {
-            switch (options.ConnectionType)
-            {
-                case MqttConnectionType.Tcp:
-                    return new MqttTcpChannel();
-                case MqttConnectionType.Ws:
-                    return new MqttWebSocketChannel();
-
-                default:
-                    throw new NotSupportedException();
-            }
         }
     }
 }

Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs

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

MQTTnet.Core/Adapter/IMqttCommunicationAdapter.cs

@@ -2,7 +2,6 @@
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
-using MQTTnet.Core.Client;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Serializer;
 
@@ -12,7 +11,7 @@ namespace MQTTnet.Core.Adapter
     {
         IMqttPacketSerializer PacketSerializer { get; }
 
-        Task ConnectAsync(TimeSpan timeout, MqttClientOptions options);
+        Task ConnectAsync(TimeSpan timeout);
 
         Task DisconnectAsync(TimeSpan timeout);
 

MQTTnet.Core/Adapter/IMqttServerAdapter.cs

@@ -1,14 +1,14 @@
 using System;
+using System.Threading.Tasks;
 using MQTTnet.Core.Server;
 
 namespace MQTTnet.Core.Adapter
 {
     public interface IMqttServerAdapter
     {
-        event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        event Action<IMqttCommunicationAdapter> ClientAccepted;
 
-        void Start(MqttServerOptions options);
+        Task StartAsync(MqttServerOptions options);
-
+        Task StopAsync();
-        void Stop();
     }
 }

MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs

@@ -4,7 +4,6 @@ using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Channel;
-using MQTTnet.Core.Client;
 using MQTTnet.Core.Diagnostics;
 using MQTTnet.Core.Exceptions;
 using MQTTnet.Core.Internal;
@@ -26,16 +25,20 @@ namespace MQTTnet.Core.Adapter
 
         public IMqttPacketSerializer PacketSerializer { get; }
 
-        public async Task ConnectAsync(TimeSpan timeout, MqttClientOptions options)
+        public async Task ConnectAsync(TimeSpan timeout)
         {
             try
             {
-                await _channel.ConnectAsync(options).TimeoutAfter(timeout).ConfigureAwait(false);
+                await _channel.ConnectAsync().TimeoutAfter(timeout).ConfigureAwait(false);
             }
             catch (TaskCanceledException)
             {
                 throw;
             }
+            catch (OperationCanceledException)
+            {
+                throw;
+            }
             catch (MqttCommunicationTimedOutException)
             {
                 throw;
@@ -60,6 +63,10 @@ namespace MQTTnet.Core.Adapter
             {
                 throw;
             }
+            catch (OperationCanceledException)
+            {
+                throw;
+            }
             catch (MqttCommunicationTimedOutException)
             {
                 throw;
@@ -87,7 +94,7 @@ namespace MQTTnet.Core.Adapter
                         continue;
                     }
 
-                    MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"TX >>> {0} [Timeout={1}]", packet, timeout);
+                    MqttNetTrace.Information(nameof(MqttChannelCommunicationAdapter), "TX >>> {0} [Timeout={1}]", packet, timeout);
 
                     var writeBuffer = PacketSerializer.Serialize(packet);
                     await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
@@ -106,6 +113,10 @@ namespace MQTTnet.Core.Adapter
             {
                 throw;
             }
+            catch (OperationCanceledException)
+            {
+                throw;
+            }
             catch (MqttCommunicationTimedOutException)
             {
                 throw;
@@ -149,13 +160,17 @@ namespace MQTTnet.Core.Adapter
                     throw new MqttProtocolViolationException("Received malformed packet.");
                 }
 
-                MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), "RX <<< {0}", packet);
+                MqttNetTrace.Information(nameof(MqttChannelCommunicationAdapter), "RX <<< {0}", packet);
                 return packet;
             }
             catch (TaskCanceledException)
             {
                 throw;
             }
+            catch (OperationCanceledException)
+            {
+                throw;
+            }
             catch (MqttCommunicationTimedOutException)
             {
                 throw;

MQTTnet.Core/Adapter/MqttClientConnectedEventArgs.cs

@@ -1,17 +1,15 @@
-using System;
+using MQTTnet.Core.Server;
+using System;
 
 namespace MQTTnet.Core.Adapter
 {
     public class MqttClientConnectedEventArgs : EventArgs
     {
-        public MqttClientConnectedEventArgs(string identifier, IMqttCommunicationAdapter clientAdapter)
+        public MqttClientConnectedEventArgs(ConnectedMqttClient client)
         {
-            Identifier = identifier ?? throw new ArgumentNullException(nameof(identifier));
+            Client = client ?? throw new ArgumentNullException(nameof(client));
-            ClientAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
         }
 
-        public string Identifier { get; }
+        public ConnectedMqttClient Client { get; }
-
-        public IMqttCommunicationAdapter ClientAdapter { get; }
     }
 }

MQTTnet.Core/Adapter/MqttClientDisconnectedEventArgs.cs

@@ -0,0 +1,15 @@
+using MQTTnet.Core.Server;
+using System;
+
+namespace MQTTnet.Core.Adapter
+{
+    public class MqttClientDisconnectedEventArgs : EventArgs
+    {
+        public MqttClientDisconnectedEventArgs(ConnectedMqttClient client)
+        {
+            Client = client ?? throw new ArgumentNullException(nameof(client));
+        }
+        
+        public ConnectedMqttClient Client { get; }
+    }
+}

MQTTnet.Core/Channel/IMqttCommunicationChannel.cs

@@ -1,5 +1,4 @@
 using System.Threading.Tasks;
-using MQTTnet.Core.Client;
 using System.IO;
 
 namespace MQTTnet.Core.Channel
@@ -10,7 +9,7 @@ namespace MQTTnet.Core.Channel
         Stream ReceiveStream { get; }
         Stream RawReceiveStream { get; }
 
-        Task ConnectAsync(MqttClientOptions options);
+        Task ConnectAsync();
         Task DisconnectAsync();
     }
 }

MQTTnet.Core/Client/IMqttClient.cs

@@ -13,7 +13,7 @@ namespace MQTTnet.Core.Client
         event EventHandler Connected;
         event EventHandler Disconnected;
 
-        Task ConnectAsync(MqttApplicationMessage willApplicationMessage = null);
+        Task ConnectAsync(MqttClientOptions options);
         Task DisconnectAsync();
 
         Task<IList<MqttSubscribeResult>> SubscribeAsync(IEnumerable<TopicFilter> topicFilters);

MQTTnet.Core/Client/IMqttClientFactory.cs

@@ -2,6 +2,6 @@
 {
     public interface IMqttClientFactory
     {
-        IMqttClient CreateMqttClient(MqttClientOptions options);
+        IMqttClient CreateMqttClient();
     }
 }

MQTTnet.Core/Client/IMqttCommunicationAdapterFactory.cs

@@ -0,0 +1,9 @@
+using MQTTnet.Core.Adapter;
+
+namespace MQTTnet.Core.Client
+{
+    public interface IMqttCommunicationAdapterFactory
+    {
+        IMqttCommunicationAdapter CreateMqttCommunicationAdapter(MqttClientOptions options);
+    }
+}

MQTTnet.Core/Client/MqttClient.cs

@@ -16,19 +16,17 @@ namespace MQTTnet.Core.Client
     {
         private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
         private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
-        private readonly MqttClientOptions _options;
+        private readonly IMqttCommunicationAdapterFactory _communicationChannelFactory;
-        private readonly IMqttCommunicationAdapter _adapter;
 
+        private MqttClientOptions _options;
         private bool _isReceivingPackets;
         private int _latestPacketIdentifier;
         private CancellationTokenSource _cancellationTokenSource;
+        private IMqttCommunicationAdapter _adapter;
 
-        public MqttClient(MqttClientOptions options, IMqttCommunicationAdapter adapter)
+        public MqttClient(IMqttCommunicationAdapterFactory communicationChannelFactory)
         {
-            _options = options ?? throw new ArgumentNullException(nameof(options));
+            _communicationChannelFactory = communicationChannelFactory ?? throw new ArgumentNullException(nameof(communicationChannelFactory));
-            _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter));
-
-            _adapter.PacketSerializer.ProtocolVersion = options.ProtocolVersion;
         }
 
         public event EventHandler Connected;
@@ -37,24 +35,29 @@ namespace MQTTnet.Core.Client
 
         public bool IsConnected => _cancellationTokenSource != null && !_cancellationTokenSource.IsCancellationRequested;
 
-        public async Task ConnectAsync(MqttApplicationMessage willApplicationMessage = null)
+        public async Task ConnectAsync(MqttClientOptions options)
         {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
             ThrowIfConnected("It is not allowed to connect with a server after the connection is established.");
 
             try
             {
+                _options = options;
                 _cancellationTokenSource = new CancellationTokenSource();
                 _latestPacketIdentifier = 0;
                 _packetDispatcher.Reset();
 
-                MqttTrace.Verbose(nameof(MqttClient), "Trying to connect with server.");
+                _adapter = _communicationChannelFactory.CreateMqttCommunicationAdapter(options);
-                await _adapter.ConnectAsync(_options.DefaultCommunicationTimeout, _options).ConfigureAwait(false);
+                
-                MqttTrace.Verbose(nameof(MqttClient), "Connection with server established.");
+                MqttNetTrace.Verbose(nameof(MqttClient), "Trying to connect with server.");
+                await _adapter.ConnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                MqttNetTrace.Verbose(nameof(MqttClient), "Connection with server established.");
 
                 await SetupIncomingPacketProcessingAsync();
-                await AuthenticateAsync(willApplicationMessage);
+                await AuthenticateAsync(options.WillMessage);
 
-                MqttTrace.Verbose(nameof(MqttClient), "MQTT connection with server established.");
+                MqttNetTrace.Verbose(nameof(MqttClient), "MQTT connection with server established.");
 
                 if (_options.KeepAlivePeriod != TimeSpan.Zero)
                 {
@@ -232,11 +235,11 @@ namespace MQTTnet.Core.Client
             try
             {
                 await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
-                MqttTrace.Information(nameof(MqttClient), "Disconnected from adapter.");
+                MqttNetTrace.Information(nameof(MqttClient), "Disconnected from adapter.");
             }
             catch (Exception exception)
             {
-                MqttTrace.Warning(nameof(MqttClient), exception, "Error while disconnecting from adapter.");
+                MqttNetTrace.Warning(nameof(MqttClient), exception, "Error while disconnecting from adapter.");
             }
             finally
             {
@@ -248,7 +251,7 @@ namespace MQTTnet.Core.Client
         {
             try
             {
-                MqttTrace.Information(nameof(MqttClient), "Received <<< {0}", packet);
+                MqttNetTrace.Information(nameof(MqttClient), "Received <<< {0}", packet);
 
                 if (packet is MqttPingReqPacket)
                 {
@@ -278,7 +281,7 @@ namespace MQTTnet.Core.Client
             }
             catch (Exception exception)
             {
-                MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while processing received packet.");
+                MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while processing received packet.");
             }
         }
 
@@ -291,7 +294,7 @@ namespace MQTTnet.Core.Client
             }
             catch (Exception exception)
             {
-                MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while handling application message.");
+                MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while handling application message.");
             }
         }
 
@@ -355,7 +358,7 @@ namespace MQTTnet.Core.Client
 
         private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
         {
-            MqttTrace.Information(nameof(MqttClient), "Start sending keep alive packets.");
+            MqttNetTrace.Information(nameof(MqttClient), "Start sending keep alive packets.");
 
             try
             {
@@ -370,7 +373,7 @@ namespace MQTTnet.Core.Client
                     await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket()).ConfigureAwait(false);
                 }
             }
-            catch (TaskCanceledException)
+            catch (OperationCanceledException)
             {
             }
             catch (MqttCommunicationException exception)
@@ -380,23 +383,23 @@ namespace MQTTnet.Core.Client
                     return;
                 }
 
-                MqttTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while sending/receiving keep alive packets.");
+                MqttNetTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while sending/receiving keep alive packets.");
                 await DisconnectInternalAsync().ConfigureAwait(false);
             }
             catch (Exception exception)
             {
-                MqttTrace.Warning(nameof(MqttClient), exception, "Unhandled exception while sending/receiving keep alive packets.");
+                MqttNetTrace.Warning(nameof(MqttClient), exception, "Unhandled exception while sending/receiving keep alive packets.");
                 await DisconnectInternalAsync().ConfigureAwait(false);
             }
             finally
             {
-                MqttTrace.Information(nameof(MqttClient), "Stopped sending keep alive packets.");
+                MqttNetTrace.Information(nameof(MqttClient), "Stopped sending keep alive packets.");
             }
         }
 
         private async Task ReceivePackets(CancellationToken cancellationToken)
         {
-            MqttTrace.Information(nameof(MqttClient), "Start receiving packets.");
+            MqttNetTrace.Information(nameof(MqttClient), "Start receiving packets.");
 
             try
             {
@@ -413,7 +416,7 @@ namespace MQTTnet.Core.Client
                     StartProcessReceivedPacket(packet, cancellationToken);
                 }
             }
-            catch (TaskCanceledException)
+            catch (OperationCanceledException)
             {
             }
             catch (MqttCommunicationException exception)
@@ -423,17 +426,17 @@ namespace MQTTnet.Core.Client
                     return;
                 }
 
-                MqttTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while receiving packets.");
+                MqttNetTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while receiving packets.");
                 await DisconnectInternalAsync().ConfigureAwait(false);
             }
             catch (Exception exception)
             {
-                MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while receiving packets.");
+                MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while receiving packets.");
                 await DisconnectInternalAsync().ConfigureAwait(false);
             }
             finally
             {
-                MqttTrace.Information(nameof(MqttClient), "Stopped receiving packets.");
+                MqttNetTrace.Information(nameof(MqttClient), "Stopped receiving packets.");
             }
         }
 

MQTTnet.Core/Client/MqttClientOptions.cs

@@ -3,14 +3,12 @@ using MQTTnet.Core.Serializer;
 
 namespace MQTTnet.Core.Client
 {
-    public sealed class MqttClientOptions
+    public abstract class MqttClientOptions
     {
-        public string Server { get; set; }
-
-        public int? Port { get; set; }
-
         public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
 
+        public MqttApplicationMessage WillMessage { get; set; }
+
         public string UserName { get; set; }
 
         public string Password { get; set; }
@@ -24,7 +22,5 @@ namespace MQTTnet.Core.Client
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
 
         public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
-
-        public MqttConnectionType ConnectionType { get; set; } = MqttConnectionType.Tcp;
     }
 }

MQTTnet.Core/Client/MqttClientTcpOptions.cs

@@ -0,0 +1,9 @@
+namespace MQTTnet.Core.Client
+{
+    public class MqttClientTcpOptions : MqttClientOptions
+    {
+        public string Server { get; set; }
+
+        public int? Port { get; set; }
+    }
+}

MQTTnet.Core/Client/MqttClientTcpOptionsExtensions.cs

@@ -2,9 +2,9 @@
 
 namespace MQTTnet.Core.Client
 {
-    public static class MqttClientOptionsExtensions
+    public static class MqttClientTcpOptionsExtensions
     {
-        public static int GetPort(this MqttClientOptions options)
+        public static int GetPort(this MqttClientTcpOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

MQTTnet.Core/Client/MqttClientTlsOptions.cs

@@ -6,7 +6,11 @@ namespace MQTTnet.Core.Client
     {
         public bool UseTls { get; set; }
 
-        public bool CheckCertificateRevocation { get; set; }
+        public bool IgnoreCertificateRevocationErrors { get; set; }
+
+        public bool IgnoreCertificateChainErrors { get; set; }
+
+        public bool AllowUntrustedCertificates { get; set; }
 
         public List<byte[]> Certificates { get; set; }
     }

MQTTnet.Core/Client/MqttClientWebSocketOptions.cs

@@ -0,0 +1,7 @@
+namespace MQTTnet.Core.Client
+{
+    public class MqttClientWebSocketOptions : MqttClientOptions
+    {
+        public string Uri { get; set; }
+    }
+}

MQTTnet.Core/Client/MqttConnectionType.cs

@@ -1,8 +0,0 @@
-namespace MQTTnet.Core.Client
-{
-    public enum MqttConnectionType
-    {
-        Tcp,
-        Ws
-    }
-}

MQTTnet.Core/Client/MqttPacketDispatcher.cs

@@ -24,7 +24,7 @@ namespace MQTTnet.Core.Client
             }
             catch (MqttCommunicationTimedOutException)
             {
-                MqttTrace.Warning(nameof(MqttPacketDispatcher), "Timeout while waiting for packet of type '{0}'.", responseType.Name);
+                MqttNetTrace.Warning(nameof(MqttPacketDispatcher), "Timeout while waiting for packet of type '{0}'.", responseType.Name);
                 throw;
             }
             finally

MQTTnet.Core/Diagnostics/MqttNetTrace.cs

@@ -2,41 +2,41 @@
 
 namespace MQTTnet.Core.Diagnostics
 {
-    public static class MqttTrace
+    public static class MqttNetTrace
     {
-        public static event EventHandler<MqttTraceMessagePublishedEventArgs> TraceMessagePublished;
+        public static event EventHandler<MqttNetTraceMessagePublishedEventArgs> TraceMessagePublished;
 
         public static void Verbose(string source, string message, params object[] parameters)
         {
-            Publish(source, MqttTraceLevel.Verbose, null, message, parameters);
+            Publish(source, MqttNetTraceLevel.Verbose, null, message, parameters);
         }
 
         public static void Information(string source, string message, params object[] parameters)
         {
-            Publish(source, MqttTraceLevel.Information, null, message, parameters);
+            Publish(source, MqttNetTraceLevel.Information, null, message, parameters);
         }
 
         public static void Warning(string source, string message, params object[] parameters)
         {
-            Publish(source, MqttTraceLevel.Warning, null, message, parameters);
+            Publish(source, MqttNetTraceLevel.Warning, null, message, parameters);
         }
 
         public static void Warning(string source, Exception exception, string message, params object[] parameters)
         {
-            Publish(source, MqttTraceLevel.Warning, exception, message, parameters);
+            Publish(source, MqttNetTraceLevel.Warning, exception, message, parameters);
         }
 
         public static void Error(string source, string message, params object[] parameters)
         {
-            Publish(source, MqttTraceLevel.Error, null, message, parameters);
+            Publish(source, MqttNetTraceLevel.Error, null, message, parameters);
         }
 
         public static void Error(string source, Exception exception, string message, params object[] parameters)
         {
-            Publish(source, MqttTraceLevel.Error, exception, message, parameters);
+            Publish(source, MqttNetTraceLevel.Error, exception, message, parameters);
         }
 
-        private static void Publish(string source, MqttTraceLevel traceLevel, Exception exception, string message, params object[] parameters)
+        private static void Publish(string source, MqttNetTraceLevel traceLevel, Exception exception, string message, params object[] parameters)
         {
             var handler = TraceMessagePublished;
             if (handler == null)
@@ -52,12 +52,12 @@ namespace MQTTnet.Core.Diagnostics
                 }
                 catch (Exception formatException)
                 {
-                    Error(nameof(MqttTrace), formatException, "Error while tracing message: " + message);
+                    Error(nameof(MqttNetTrace), formatException, "Error while tracing message: " + message);
                     return;
                 }
             }
 
-            handler.Invoke(null, new MqttTraceMessagePublishedEventArgs(Environment.CurrentManagedThreadId, source, traceLevel, message, exception));
+            handler.Invoke(null, new MqttNetTraceMessagePublishedEventArgs(Environment.CurrentManagedThreadId, source, traceLevel, message, exception));
         }
     }
 }

MQTTnet.Core/Diagnostics/MqttNetTraceLevel.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Diagnostics
 {
-    public enum MqttTraceLevel
+    public enum MqttNetTraceLevel
     {
         Verbose,
         Information,

MQTTnet.Core/Diagnostics/MqttNetTraceMessagePublishedEventArgs.cs

@@ -2,9 +2,9 @@
 
 namespace MQTTnet.Core.Diagnostics
 {
-    public sealed class MqttTraceMessagePublishedEventArgs : EventArgs
+    public sealed class MqttNetTraceMessagePublishedEventArgs : EventArgs
     {
-        public MqttTraceMessagePublishedEventArgs(int threadId, string source, MqttTraceLevel level, string message, Exception exception)
+        public MqttNetTraceMessagePublishedEventArgs(int threadId, string source, MqttNetTraceLevel level, string message, Exception exception)
         {
             ThreadId = threadId;
             Source = source;
@@ -17,7 +17,7 @@ namespace MQTTnet.Core.Diagnostics
 
         public string Source { get; }
 
-        public MqttTraceLevel Level { get; }
+        public MqttNetTraceLevel Level { get; }
 
         public string Message { get; }
 

MQTTnet.Core/MQTTnet.Core.csproj

@@ -17,8 +17,8 @@
     <PackageIconUrl></PackageIconUrl>
     <RepositoryUrl></RepositoryUrl>
     <PackageTags></PackageTags>
-    <FileVersion>2.3.1.0</FileVersion>
+    <FileVersion>2.4.0.0</FileVersion>
-    <AssemblyVersion>2.3.1.0</AssemblyVersion>
+    <AssemblyVersion>2.4.0.0</AssemblyVersion>
     <PackageLicenseUrl></PackageLicenseUrl>
   </PropertyGroup>
 

MQTTnet.Core/Packets/MqttSubscribePacket.cs

@@ -11,7 +11,7 @@ namespace MQTTnet.Core.Packets
 
         public override string ToString()
         {
-            var topicFiltersText = string.Join(",", TopicFilters.Select(f => $"{f.Topic}@{f.QualityOfServiceLevel}"));
+            var topicFiltersText = string.Join(",", TopicFilters.Select(f => f.Topic + "@" + f.QualityOfServiceLevel));
             return nameof(MqttSubscribePacket) + ": [PacketIdentifier=" + PacketIdentifier + "] [TopicFilters=" + topicFiltersText + "]";
         }
     }

MQTTnet.Core/Serializer/MqttPacketSerializer.cs

@@ -212,8 +212,7 @@ namespace MQTTnet.Core.Serializer
 
                 default:
                     {
-                        throw new MqttProtocolViolationException(
+                        throw new MqttProtocolViolationException($"Packet type ({(int)header.ControlPacketType}) not supported.");
-                            $"Packet type ({(int)header.ControlPacketType}) not supported.");
                     }
             }
         }

MQTTnet.Core/Server/IMqttServer.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 
 namespace MQTTnet.Core.Server
@@ -8,11 +9,12 @@ namespace MQTTnet.Core.Server
     {
         event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
         event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
 
-        IList<ConnectedMqttClient> GetConnectedClients();
+        IReadOnlyList<ConnectedMqttClient> GetConnectedClients();
-        void InjectClient(string identifier, IMqttCommunicationAdapter adapter);
         void Publish(MqttApplicationMessage applicationMessage);
-        void Start();
+
-        void Stop();
+        Task StartAsync();
+        Task StopAsync();
     }
 }

MQTTnet.Core/Server/IMqttServerStorage.cs

@@ -0,0 +1,12 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Core.Server
+{
+    public interface IMqttServerStorage
+    {
+        Task SaveRetainedMessagesAsync(IList<MqttApplicationMessage> messages);
+
+        Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync();
+    }
+}

MQTTnet.Core/Server/MqttClientMessageQueue.cs

@@ -1,89 +0,0 @@
-using System;
-using System.Collections.Concurrent;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Diagnostics;
-using MQTTnet.Core.Exceptions;
-using MQTTnet.Core.Packets;
-using System.Linq;
-
-namespace MQTTnet.Core.Server
-{
-    public sealed class MqttClientMessageQueue
-    {
-        private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>();
-
-        private readonly MqttServerOptions _options;
-        private CancellationTokenSource _cancellationTokenSource;
-
-        public MqttClientMessageQueue(MqttServerOptions options)
-        {
-            _options = options ?? throw new ArgumentNullException(nameof(options));
-        }
-
-        public void Start(IMqttCommunicationAdapter adapter)
-        {
-            if (_cancellationTokenSource != null)
-            {
-                throw new InvalidOperationException($"{nameof(MqttClientMessageQueue)} already started.");
-            }
-
-            if (adapter == null) throw new ArgumentNullException(nameof(adapter));
-            _cancellationTokenSource = new CancellationTokenSource();
-
-            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token, adapter), _cancellationTokenSource.Token);
-        }
-
-        public void Stop()
-        {
-            _cancellationTokenSource?.Cancel();
-            _cancellationTokenSource = null;
-            _pendingPublishPackets?.Dispose();
-        }
-
-        public void Enqueue(MqttPublishPacket publishPacket)
-        {
-            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
-
-            _pendingPublishPackets.Add(publishPacket);
-        }
-
-        private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken, IMqttCommunicationAdapter adapter)
-        {
-            var consumable = _pendingPublishPackets.GetConsumingEnumerable();
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                if (_pendingPublishPackets.Count == 0)
-                {
-                    await Task.Delay(TimeSpan.FromMilliseconds(5), cancellationToken).ConfigureAwait(false);
-                    continue;
-                }
-
-                var packets = consumable.Take(_pendingPublishPackets.Count).ToList();
-                try
-                {
-                    await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packets).ConfigureAwait(false);
-                }
-                catch (MqttCommunicationException exception)
-                {
-                    MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
-                    foreach (var publishPacket in packets)
-                    {
-                        publishPacket.Dup = true;
-                        _pendingPublishPackets.Add(publishPacket, cancellationToken);
-                    }
-                }
-                catch (Exception exception)
-                {
-                    MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
-                    foreach (var publishPacket in packets)
-                    {
-                        publishPacket.Dup = true;
-                        _pendingPublishPackets.Add(publishPacket, cancellationToken);
-                    }
-                }
-            }
-        }
-    }
-}

MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs

@@ -0,0 +1,93 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Core.Adapter;
+using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.Exceptions;
+using MQTTnet.Core.Packets;
+using MQTTnet.Core.Protocol;
+
+namespace MQTTnet.Core.Server
+{
+    public sealed class MqttClientPendingMessagesQueue
+    {
+        private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>();
+        private readonly MqttClientSession _session;
+        private readonly MqttServerOptions _options;
+
+        public MqttClientPendingMessagesQueue(MqttServerOptions options, MqttClientSession session)
+        {
+            _session = session ?? throw new ArgumentNullException(nameof(session));
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
+        public void Start(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            if (adapter == null) throw new ArgumentNullException(nameof(adapter));
+
+            Task.Run(() => SendPendingPublishPacketsAsync(adapter, cancellationToken), cancellationToken);
+        }
+
+        public void Enqueue(MqttPublishPacket publishPacket)
+        {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
+            _pendingPublishPackets.Add(publishPacket);
+        }
+
+        private async Task SendPendingPublishPacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    await SendPendingPublishPacketAsync(adapter, cancellationToken);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (Exception exception)
+            {
+                MqttNetTrace.Error(nameof(MqttClientPendingMessagesQueue), exception, "Unhandled exception while sending pending publish packets.");
+            }
+        }
+
+        private async Task SendPendingPublishPacketAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            var packet = _pendingPublishPackets.Take(cancellationToken);
+
+            try
+            {
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                if (exception is MqttCommunicationTimedOutException)
+                {
+                    MqttNetTrace.Warning(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed due to timeout.");
+                }
+                else if (exception is MqttCommunicationException)
+                {
+                    MqttNetTrace.Warning(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed due to communication exception.");
+                }
+                if (exception is OperationCanceledException)
+                {
+                }
+                else
+                {
+                    MqttNetTrace.Error(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed.");
+                }
+
+                if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+                {
+                    packet.Dup = true;
+                    _pendingPublishPackets.Add(packet, cancellationToken);
+                }
+
+                _session.Stop();
+            }
+        }
+    }
+}

MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs

@@ -0,0 +1,109 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.Internal;
+using MQTTnet.Core.Packets;
+
+namespace MQTTnet.Core.Server
+{
+    public sealed class MqttClientRetainedMessagesManager
+    {
+        private readonly Dictionary<string, MqttPublishPacket> _retainedMessages = new Dictionary<string, MqttPublishPacket>();
+        private readonly MqttServerOptions _options;
+
+        public MqttClientRetainedMessagesManager(MqttServerOptions options)
+        {
+            _options = options ?? throw new ArgumentNullException(nameof(options));
+        }
+
+        public async Task LoadMessagesAsync()
+        {
+            if (_options.Storage == null)
+            {
+                return;
+            }
+
+            try
+            {
+                var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync();
+                lock (_retainedMessages)
+                {
+                    _retainedMessages.Clear();
+                    foreach (var retainedMessage in retainedMessages)
+                    {
+                        _retainedMessages[retainedMessage.Topic] = retainedMessage.ToPublishPacket();
+                    }
+                }
+            }
+            catch (Exception exception)
+            {
+                MqttNetTrace.Error(nameof(MqttClientRetainedMessagesManager), exception, "Unhandled exception while loading retained messages.");
+            }
+        }
+
+        public async Task HandleMessageAsync(string clientId, MqttPublishPacket publishPacket)
+        {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
+            List<MqttPublishPacket> allRetainedMessages;
+            lock (_retainedMessages)
+            {
+                if (publishPacket.Payload?.Any() == false)
+                {
+                    _retainedMessages.Remove(publishPacket.Topic);
+                    MqttNetTrace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' cleared retained message for topic '{1}'.", clientId, publishPacket.Topic);
+                }
+                else
+                {
+                    _retainedMessages[publishPacket.Topic] = publishPacket;
+                    MqttNetTrace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' updated retained message for topic '{1}'.", clientId, publishPacket.Topic);
+                }
+
+                allRetainedMessages = new List<MqttPublishPacket>(_retainedMessages.Values);
+            }
+
+            try
+            {
+                // ReSharper disable once UseNullPropagation
+                if (_options.Storage != null)
+                {
+                    await _options.Storage.SaveRetainedMessagesAsync(allRetainedMessages.Select(p => p.ToApplicationMessage()).ToList());
+                }
+            }
+            catch (Exception exception)
+            {
+                MqttNetTrace.Error(nameof(MqttClientRetainedMessagesManager), exception, "Unhandled exception while saving retained messages.");
+            }
+        }
+
+        public List<MqttPublishPacket> GetMessages(MqttSubscribePacket subscribePacket)
+        {
+            var retainedMessages = new List<MqttPublishPacket>();
+            lock (_retainedMessages)
+            {
+                foreach (var retainedMessage in _retainedMessages.Values)
+                {
+                    foreach (var topicFilter in subscribePacket.TopicFilters)
+                    {
+                        if (retainedMessage.QualityOfServiceLevel < topicFilter.QualityOfServiceLevel)
+                        {
+                            continue;
+                        }
+
+                        if (!MqttTopicFilterComparer.IsMatch(retainedMessage.Topic, topicFilter.Topic))
+                        {
+                            continue;
+                        }
+
+                        retainedMessages.Add(retainedMessage);
+                        break;
+                    }
+                }
+            }
+
+            return retainedMessages;
+        }
+    }
+}

MQTTnet.Core/Server/MqttClientSession.cs

@@ -8,76 +8,78 @@ using MQTTnet.Core.Exceptions;
 using MQTTnet.Core.Internal;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Serializer;
 
 namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSession : IDisposable
     {
         private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
-
+        
         private readonly MqttClientSubscriptionsManager _subscriptionsManager = new MqttClientSubscriptionsManager();
-        private readonly MqttClientMessageQueue _messageQueue;
+        private readonly MqttClientSessionsManager _mqttClientSessionsManager;
-        private readonly Action<MqttClientSession, MqttPublishPacket> _publishPacketReceivedCallback;
+        private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
         private readonly MqttServerOptions _options;
 
+        private IMqttCommunicationAdapter _adapter;
         private CancellationTokenSource _cancellationTokenSource;
-        private string _identifier;
+        private MqttApplicationMessage _willMessage;
-        private MqttApplicationMessage _willApplicationMessage;
 
-        public MqttClientSession(string clientId, MqttServerOptions options, Action<MqttClientSession, MqttPublishPacket> publishPacketReceivedCallback)
+        public MqttClientSession(string clientId, MqttServerOptions options, MqttClientSessionsManager mqttClientSessionsManager)
         {
             ClientId = clientId;
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            _publishPacketReceivedCallback = publishPacketReceivedCallback ?? throw new ArgumentNullException(nameof(publishPacketReceivedCallback));
+            _mqttClientSessionsManager = mqttClientSessionsManager ?? throw new ArgumentNullException(nameof(mqttClientSessionsManager));
-
+            _pendingMessagesQueue = new MqttClientPendingMessagesQueue(options, this);
-            _messageQueue = new MqttClientMessageQueue(options);
         }
 
         public string ClientId { get; }
 
-        public bool IsConnected => Adapter != null;
+        public MqttProtocolVersion? ProtocolVersion => _adapter?.PacketSerializer.ProtocolVersion;
 
-        public IMqttCommunicationAdapter Adapter { get; private set; }
+        public bool IsConnected => _adapter != null;
 
-        public async Task RunAsync(string identifier, MqttApplicationMessage willApplicationMessage, IMqttCommunicationAdapter adapter)
+        public async Task RunAsync(MqttApplicationMessage willMessage, IMqttCommunicationAdapter adapter)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));
 
-            _willApplicationMessage = willApplicationMessage;
+            _willMessage = willMessage;
 
             try
             {
-                _identifier = identifier;
+                _adapter = adapter;
-                Adapter = adapter;
                 _cancellationTokenSource = new CancellationTokenSource();
 
-                _messageQueue.Start(adapter);
+                _pendingMessagesQueue.Start(adapter, _cancellationTokenSource.Token);
-                while (!_cancellationTokenSource.IsCancellationRequested)
+                await ReceivePacketsAsync(adapter, _cancellationTokenSource.Token);
-                {
-                    var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false);
-                    await HandleIncomingPacketAsync(packet).ConfigureAwait(false);
-                }
             }
-            catch (MqttCommunicationException)
+            catch (OperationCanceledException)
             {
             }
+            catch (MqttCommunicationException exception)
+            {
+                MqttNetTrace.Warning(nameof(MqttClientSession), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
+            }
             catch (Exception exception)
             {
-                MqttTrace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", _identifier);
+                MqttNetTrace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
             }
-            finally
+        }
+
+        public void Stop()
+        {
+            if (_willMessage != null)
             {
-                if (willApplicationMessage != null)
+                _mqttClientSessionsManager.DispatchPublishPacket(this, _willMessage.ToPublishPacket());
-                {
-                    _publishPacketReceivedCallback(this, _willApplicationMessage.ToPublishPacket());
-                }
-
-                _messageQueue.Stop();
-                _cancellationTokenSource.Cancel();
-                Adapter = null;
-
-                MqttTrace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", _identifier);
             }
+
+            _cancellationTokenSource?.Cancel(false);
+            _cancellationTokenSource?.Dispose();
+            _cancellationTokenSource = null;
+
+            _adapter = null;
+
+            MqttNetTrace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", ClientId);
         }
 
         public void EnqueuePublishPacket(MqttPublishPacket publishPacket)
@@ -89,8 +91,8 @@ namespace MQTTnet.Core.Server
                 return;
             }
 
-            _messageQueue.Enqueue(publishPacket);
+            _pendingMessagesQueue.Enqueue(publishPacket);
-            MqttTrace.Verbose(nameof(MqttClientSession), "Client '{0}': Enqueued pending publish packet.", _identifier);
+            MqttNetTrace.Verbose(nameof(MqttClientSession), "Client '{0}': Enqueued pending publish packet.", ClientId);
         }
 
         public void Dispose()
@@ -99,68 +101,100 @@ namespace MQTTnet.Core.Server
             _cancellationTokenSource?.Dispose();
         }
 
-        private Task HandleIncomingPacketAsync(MqttBasePacket packet)
+        private async Task ReceivePacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
+                    await ProcessReceivedPacketAsync(adapter, packet).ConfigureAwait(false);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (MqttCommunicationException exception)
+            {
+                MqttNetTrace.Warning(nameof(MqttClientSession), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
+                Stop();
+            }
+            catch (Exception exception)
+            {
+                MqttNetTrace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
+                Stop();
+            }
+        }
+
+        private async Task ProcessReceivedPacketAsync(IMqttCommunicationAdapter adapter, MqttBasePacket packet)
         {
             if (packet is MqttSubscribePacket subscribePacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Subscribe(subscribePacket));
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Subscribe(subscribePacket));
+                EnqueueRetainedMessages(subscribePacket);
             }
-
+            else if (packet is MqttUnsubscribePacket unsubscribePacket)
-            if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Unsubscribe(unsubscribePacket));
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, _subscriptionsManager.Unsubscribe(unsubscribePacket));
             }
-
+            else if (packet is MqttPublishPacket publishPacket)
-            if (packet is MqttPublishPacket publishPacket)
             {
-                return HandleIncomingPublishPacketAsync(publishPacket);
+                await HandleIncomingPublishPacketAsync(adapter, publishPacket);
             }
-
+            else if (packet is MqttPubRelPacket pubRelPacket)
-            if (packet is MqttPubRelPacket pubRelPacket)
             {
-                return HandleIncomingPubRelPacketAsync(pubRelPacket);
+                await HandleIncomingPubRelPacketAsync(adapter, pubRelPacket);
             }
-
+            else if (packet is MqttPubRecPacket pubRecPacket)
-            if (packet is MqttPubRecPacket pubRecPacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, pubRecPacket.CreateResponse<MqttPubRelPacket>());
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, pubRecPacket.CreateResponse<MqttPubRelPacket>());
             }
-
+            else if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
-            if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
             {
                 // Discard message.
-                return Task.FromResult((object)null);
             }
-
+            else if (packet is MqttPingReqPacket)
-            if (packet is MqttPingReqPacket)
             {
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPingRespPacket());
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPingRespPacket());
             }
-
+            else if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
-            if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
             {
-                _cancellationTokenSource.Cancel();
+                Stop();
-                return Task.FromResult((object)null);
+            }
+            else
+            {
+                MqttNetTrace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
+                Stop();
             }
-
-            MqttTrace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", _identifier, packet);
-            _cancellationTokenSource.Cancel();
-
-            return Task.FromResult((object)null);
         }
 
-        private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
+        private void EnqueueRetainedMessages(MqttSubscribePacket subscribePacket)
         {
+            var retainedMessages = _mqttClientSessionsManager.RetainedMessagesManager.GetMessages(subscribePacket);
+            foreach (var publishPacket in retainedMessages)
+            {
+                EnqueuePublishPacket(publishPacket);
+            }
+        }
+
+        private async Task HandleIncomingPublishPacketAsync(IMqttCommunicationAdapter adapter, MqttPublishPacket publishPacket)
+        {
+            if (publishPacket.Retain)
+            {
+                await _mqttClientSessionsManager.RetainedMessagesManager.HandleMessageAsync(ClientId, publishPacket);
+            }
+
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
-                _publishPacketReceivedCallback(this, publishPacket);
+                _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket);
-                return Task.FromResult(0);
+                return;
             }
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
-                _publishPacketReceivedCallback(this, publishPacket);
+                _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket);
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                return;
             }
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -171,22 +205,23 @@ namespace MQTTnet.Core.Server
                     _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
                 }
 
-                _publishPacketReceivedCallback(this, publishPacket);
+                _mqttClientSessionsManager.DispatchPublishPacket(this, publishPacket);
 
-                return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
+                return;
             }
 
             throw new MqttCommunicationException("Received a not supported QoS level.");
         }
 
-        private Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket)
+        private Task HandleIncomingPubRelPacketAsync(IMqttCommunicationAdapter adapter, MqttPubRelPacket pubRelPacket)
         {
             lock (_unacknowledgedPublishPackets)
             {
                 _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
             }
 
-            return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier });
+            return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _cancellationTokenSource.Token, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier });
         }
     }
 }

MQTTnet.Core/Server/MqttClientSessionsManager.cs

@@ -9,38 +9,46 @@ using MQTTnet.Core.Exceptions;
 using MQTTnet.Core.Internal;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Serializer;
 
 namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSessionsManager
     {
-        private readonly object _syncRoot = new object();
         private readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>();
         private readonly MqttServerOptions _options;
 
         public MqttClientSessionsManager(MqttServerOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+            RetainedMessagesManager = new MqttClientRetainedMessagesManager(options);
         }
 
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
 
-        public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
+        public MqttClientRetainedMessagesManager RetainedMessagesManager { get; }
+
+        public async Task RunClientSessionAsync(IMqttCommunicationAdapter clientAdapter)
         {
+            var clientId = string.Empty;
             try
             {
-                if (!(await eventArgs.ClientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false) is MqttConnectPacket connectPacket))
+                if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false) is MqttConnectPacket connectPacket))
                 {
                     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.
+                clientId = connectPacket.ClientId;
-                eventArgs.ClientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion;
 
+                // Switch to the required protocol version before sending any response.
+                clientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion;
+                
                 var connectReturnCode = ValidateConnection(connectPacket);
                 if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
                 {
-                    await eventArgs.ClientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
+                    await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
                     {
                         ConnectReturnCode = connectReturnCode
                     }).ConfigureAwait(false);
@@ -50,44 +58,84 @@ namespace MQTTnet.Core.Server
 
                 var clientSession = GetOrCreateClientSession(connectPacket);
 
-                await eventArgs.ClientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
+                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
                 {
                     ConnectReturnCode = connectReturnCode,
                     IsSessionPresent = clientSession.IsExistingSession
                 }).ConfigureAwait(false);
 
-                await clientSession.Session.RunAsync(eventArgs.Identifier, connectPacket.WillMessage, eventArgs.ClientAdapter).ConfigureAwait(false);
+                ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(new ConnectedMqttClient
+                {
+                    ClientId = clientId,
+                    ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
+                }));
+
+                await clientSession.Session.RunAsync(connectPacket.WillMessage, clientAdapter).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
-                MqttTrace.Error(nameof(MqttServer), exception, exception.Message);
+                MqttNetTrace.Error(nameof(MqttServer), exception, exception.Message);
             }
             finally
             {
-                await eventArgs.ClientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                try
+                {
+                    await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                }
+                catch (Exception)
+                {
+                    //ignored
+                }
+
+                ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(new ConnectedMqttClient
+                {
+                    ClientId = clientId,
+                    ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
+                }));
             }
         }
 
         public void Clear()
         {
-            lock (_syncRoot)
+            lock (_clientSessions)
             {
                 _clientSessions.Clear();
             }
         }
 
-        public IList<ConnectedMqttClient> GetConnectedClients()
+        public IReadOnlyList<ConnectedMqttClient> GetConnectedClients()
         {
-            lock (_syncRoot)
+            lock (_clientSessions)
             {
                 return _clientSessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
                 {
                     ClientId = s.Value.ClientId,
-                    ProtocolVersion = s.Value.Adapter.PacketSerializer.ProtocolVersion
+                    ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311
                 }).ToList();
             }
         }
 
+        public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        {
+            try
+            {
+                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, publishPacket.ToApplicationMessage());
+                ApplicationMessageReceived?.Invoke(this, eventArgs);
+            }
+            catch (Exception exception)
+            {
+                MqttNetTrace.Error(nameof(MqttClientSessionsManager), exception, "Error while processing application message");
+            }
+
+            lock (_clientSessions)
+            {
+                foreach (var clientSession in _clientSessions.Values.ToList())
+                {
+                    clientSession.EnqueuePublishPacket(publishPacket);
+                }
+            }
+        }
+
         private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
         {
             if (_options.ConnectionValidator != null)
@@ -100,10 +148,9 @@ namespace MQTTnet.Core.Server
 
         private GetOrCreateClientSessionResult GetOrCreateClientSession(MqttConnectPacket connectPacket)
         {
-            lock (_syncRoot)
+            lock (_clientSessions)
             {
-                MqttClientSession clientSession;
+                var isSessionPresent = _clientSessions.TryGetValue(connectPacket.ClientId, out var clientSession);
-                var isSessionPresent = _clientSessions.TryGetValue(connectPacket.ClientId, out clientSession);
                 if (isSessionPresent)
                 {
                     if (connectPacket.CleanSession)
@@ -111,11 +158,11 @@ namespace MQTTnet.Core.Server
                         _clientSessions.Remove(connectPacket.ClientId);
                         clientSession.Dispose();
                         clientSession = null;
-                        MqttTrace.Verbose(nameof(MqttClientSessionsManager), "Disposed existing session of client '{0}'.", connectPacket.ClientId);
+                        MqttNetTrace.Verbose(nameof(MqttClientSessionsManager), "Disposed existing session of client '{0}'.", connectPacket.ClientId);
                     }
                     else
                     {
-                        MqttTrace.Verbose(nameof(MqttClientSessionsManager), "Reusing existing session of client '{0}'.", connectPacket.ClientId);
+                        MqttNetTrace.Verbose(nameof(MqttClientSessionsManager), "Reusing existing session of client '{0}'.", connectPacket.ClientId);
                     }
                 }
 
@@ -124,35 +171,14 @@ namespace MQTTnet.Core.Server
                 {
                     isExistingSession = false;
 
-                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, DispatchPublishPacket);
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, this);
                     _clientSessions[connectPacket.ClientId] = clientSession;
 
-                    MqttTrace.Verbose(nameof(MqttClientSessionsManager), "Created a new session for client '{0}'.", connectPacket.ClientId);
+                    MqttNetTrace.Verbose(nameof(MqttClientSessionsManager), "Created a new session for client '{0}'.", connectPacket.ClientId);
                 }
 
                 return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
             }
         }
-
-        public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
-        {
-            try
-            {
-                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, publishPacket.ToApplicationMessage());
-                ApplicationMessageReceived?.Invoke(this, eventArgs);
-            }
-            catch (Exception exception)
-            {
-                MqttTrace.Error(nameof(MqttClientSessionsManager), exception, "Error while processing application message");
-            }
-
-            lock (_syncRoot)
-            {
-                foreach (var clientSession in _clientSessions.Values.ToList())
-                {
-                    clientSession.EnqueuePublishPacket(publishPacket);
-                }
-            }
-        }
     }
 }

MQTTnet.Core/Server/MqttServer.cs

@@ -23,15 +23,17 @@ namespace MQTTnet.Core.Server
 
             _clientSessionsManager = new MqttClientSessionsManager(options);
             _clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
+            _clientSessionsManager.ClientConnected += OnClientConnected;
+            _clientSessionsManager.ClientDisconnected += OnClientDisconnected;
         }
 
-        public IList<ConnectedMqttClient> GetConnectedClients()
+        public IReadOnlyList<ConnectedMqttClient> GetConnectedClients()
         {
             return _clientSessionsManager.GetConnectedClients();
         }
 
         public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
-
+        public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
         public void Publish(MqttApplicationMessage applicationMessage)
@@ -41,31 +43,24 @@ namespace MQTTnet.Core.Server
             _clientSessionsManager.DispatchPublishPacket(null, applicationMessage.ToPublishPacket());
         }
 
-        public void InjectClient(string identifier, IMqttCommunicationAdapter adapter)
+        public async Task StartAsync()
-        {
-            if (adapter == null) throw new ArgumentNullException(nameof(adapter));
-
-            if (_cancellationTokenSource == null) throw new InvalidOperationException("The MQTT server is not started.");
-
-            OnClientConnected(this, new MqttClientConnectedEventArgs(identifier, adapter));
-        }
-
-        public void Start()
         {
             if (_cancellationTokenSource != null) throw new InvalidOperationException("The MQTT server is already started.");
 
             _cancellationTokenSource = new CancellationTokenSource();
 
+            await _clientSessionsManager.RetainedMessagesManager.LoadMessagesAsync();
+
             foreach (var adapter in _adapters)
             {
-                adapter.ClientConnected += OnClientConnected;
+                adapter.ClientAccepted += OnClientAccepted;
-                adapter.Start(_options);
+                await adapter.StartAsync(_options);
             }
 
-            MqttTrace.Information(nameof(MqttServer), "Started.");
+            MqttNetTrace.Information(nameof(MqttServer), "Started.");
         }
 
-        public void Stop()
+        public async Task StopAsync()
         {
             _cancellationTokenSource?.Cancel(false);
             _cancellationTokenSource?.Dispose();
@@ -73,21 +68,30 @@ namespace MQTTnet.Core.Server
 
             foreach (var adapter in _adapters)
             {
-                adapter.ClientConnected -= OnClientConnected;
+                adapter.ClientAccepted -= OnClientAccepted;
-                adapter.Stop();
+                await adapter.StopAsync();
             }
 
             _clientSessionsManager.Clear();
 
-            MqttTrace.Information(nameof(MqttServer), "Stopped.");
+            MqttNetTrace.Information(nameof(MqttServer), "Stopped.");
+        }
+
+        private void OnClientAccepted(IMqttCommunicationAdapter adapter)
+        {
+            Task.Run(() =>_clientSessionsManager.RunClientSessionAsync(adapter), _cancellationTokenSource.Token);
         }
 
         private void OnClientConnected(object sender, MqttClientConnectedEventArgs eventArgs)
         {
-            MqttTrace.Information(nameof(MqttServer), "Client '{0}': Connected.", eventArgs.Identifier);
+            MqttNetTrace.Information(nameof(MqttServer), "Client '{0}': Connected.", eventArgs.Client.ClientId);
             ClientConnected?.Invoke(this, eventArgs);
+        }
 
-            Task.Run(() => _clientSessionsManager.RunClientSessionAsync(eventArgs), _cancellationTokenSource.Token);
+        private void OnClientDisconnected(object sender, MqttClientDisconnectedEventArgs eventArgs)
+        {
+            MqttNetTrace.Information(nameof(MqttServer), "Client '{0}': Disconnected.", eventArgs.Client.ClientId);
+            ClientDisconnected?.Invoke(this, eventArgs);
         }
     }
 }

MQTTnet.Core/Server/MqttServerOptions.cs

@@ -15,5 +15,7 @@ namespace MQTTnet.Core.Server
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
 
         public Func<MqttConnectPacket, MqttConnectReturnCode> ConnectionValidator { get; set; }
+
+        public IMqttServerStorage Storage { get; set; }
     }
 }

MQTTnet.sln

@@ -3,7 +3,7 @@ Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio 15
 VisualStudioVersion = 15.0.26730.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}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Core.Tests", "Tests\MQTTnet.Core.Tests\MQTTnet.Core.Tests.csproj", "{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.NetFramework", "Frameworks\MQTTnet.NetFramework\MQTTnet.NetFramework.csproj", "{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}"
 EndProject
@@ -50,18 +50,18 @@ Global
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|ARM.ActiveCfg = Debug|Any CPU
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|ARM.Build.0 = Debug|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.ActiveCfg = Debug|x64
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.ActiveCfg = Debug|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.Build.0 = Debug|x64
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x64.Build.0 = Debug|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.ActiveCfg = Debug|x86
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.ActiveCfg = Debug|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.Build.0 = Debug|x86
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Debug|x86.Build.0 = Debug|Any CPU
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|Any CPU.Build.0 = Release|Any CPU
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|ARM.ActiveCfg = Release|Any CPU
 		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|ARM.Build.0 = Release|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.ActiveCfg = Release|x64
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.ActiveCfg = Release|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.Build.0 = Release|x64
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x64.Build.0 = Release|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.ActiveCfg = Release|x86
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.ActiveCfg = Release|Any CPU
-		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.Build.0 = Release|x86
+		{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}.Release|x86.Build.0 = Release|Any CPU
 		{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{A480EF90-0EAA-4D9A-B271-47A9C47F6F7D}.Debug|ARM.ActiveCfg = Debug|Any CPU

README.md

@@ -18,7 +18,7 @@ MQTTnet is a .NET library for MQTT based communication. It provides a MQTT clien
 * Performance optimized (processing ~27.000 messages / second)*
 * Interfaces included for mocking and testing
 * Access to internal trace messages
-* Unit tested (57+ tests)
+* Unit tested (62+ tests)
 
 \* Tested on local machine with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetFramework_.
 
@@ -32,6 +32,7 @@ MQTTnet is a .NET library for MQTT based communication. It provides a MQTT clien
 * Able to publish its own messages (no loopback client required)
 * Able to receive every messages (no loopback client required)
 * Extensible client credential validation
+* Retained messages are supported including persisting via interface methods (own implementation required)
 
 # Supported frameworks
 * .NET Standard 1.3+
@@ -39,6 +40,7 @@ MQTTnet is a .NET library for MQTT based communication. It provides a MQTT clien
 * .NET Core App 1.1+
 * .NET Framework 4.5.2+ (x86, x64, AnyCPU)
 * Universal Windows (UWP) 10.0.10240+ (x86, x64, ARM, AnyCPU)
+* Mono 5.2+
 
 # Supported MQTT versions
 * 3.1.1
@@ -47,6 +49,9 @@ MQTTnet is a .NET library for MQTT based communication. It provides a MQTT clien
 # Nuget
 This library is available as a nuget package: https://www.nuget.org/packages/MQTTnet/
 
+# Examples
+Please find examples and the documentation at the Wiki of this repository (https://github.com/chkr1011/MQTTnet/wiki).
+
 # Contributions
 If you want to contribute to this project just create a pull request.
 
@@ -56,98 +61,4 @@ 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.
+If you use this library and want to see your project here please let me know.
-
-# Examples
-## MqttClient
-
-```csharp
-var options = new MqttClientOptions
-{
-    Server = "localhost"
-};
-
-var client = new MqttClientFactory().CreateMqttClient(options);
-client.ApplicationMessageReceived += (s, e) =>
-{
-    Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
-    Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
-    Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
-    Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
-    Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");
-    Console.WriteLine();
-};
-
-client.Connected += async (s, e) =>
-{
-    Console.WriteLine("### CONNECTED WITH SERVER, SUBSCRIBING ###");
-
-    await client.SubscribeAsync(new List<TopicFilter>
-    {
-        new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)
-    });
-};
-
-client.Disconnected += async (s, e) => 
-{
-    Console.WriteLine("### DISCONNECTED FROM SERVER ###");
-    await Task.Delay(TimeSpan.FromSeconds(5));
-
-    try
-    {
-        await client.ConnectAsync();
-    }
-    catch
-    {
-        Console.WriteLine("### RECONNECTING FAILED ###");
-    }
-};
-
-try
-{
-    await client.ConnectAsync();
-}
-catch
-{
-    Console.WriteLine("### CONNECTING FAILED ###");
-}
-
-Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###");
-
-var messageFactory = new MqttApplicationMessageFactory();
-while (true)
-{
-    Console.ReadLine();
-
-    var applicationMessage = messageFactory.CreateApplicationMessage("myTopic", "Hello World", MqttQualityOfServiceLevel.AtLeastOnce);
-    await client.PublishAsync(applicationMessage);
-}
-```
-
-## MqttServer
-
-```csharp
-var options = new MqttServerOptions
-{
-    ConnectionValidator = p =>
-    {
-        if (p.ClientId == "SpecialClient")
-        {
-            if (p.Username != "USER" || p.Password != "PASS")
-            {
-                return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
-            }
-        }
-
-        return MqttConnectReturnCode.ConnectionAccepted;
-    }
-};
-
-var mqttServer = new MqttServerFactory().CreateMqttServer(options);
-mqttServer.Start();
-
-Console.WriteLine("Press any key to exit.");
-Console.ReadLine();
-
-mqttServer.Stop();
-```

Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj

@@ -1,115 +1,20 @@
-<?xml version="1.0" encoding="utf-8"?>
+<Project Sdk="Microsoft.NET.Sdk">
-<Project ToolsVersion="14.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+
-  <Import Project="$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props" Condition="Exists('$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props')" />
   <PropertyGroup>
-    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
+    <OutputType>Exe</OutputType>
-    <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
+    <DebugType>Full</DebugType>
-    <ProjectGuid>{A7FF0C91-25DE-4BA6-B39E-F54E8DADF1CC}</ProjectGuid>
+    <TargetFramework>netcoreapp2.0</TargetFramework>
-    <OutputType>Library</OutputType>
-    <AppDesignerFolder>Properties</AppDesignerFolder>
-    <RootNamespace>MQTTnet.Core.Tests</RootNamespace>
-    <AssemblyName>MQTTnet.Core.Tests</AssemblyName>
-    <TargetFrameworkVersion>v4.5.2</TargetFrameworkVersion>
-    <FileAlignment>512</FileAlignment>
-    <TargetFrameworkProfile />
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
-    <DebugSymbols>true</DebugSymbols>
-    <DebugType>full</DebugType>
-    <Optimize>false</Optimize>
-    <OutputPath>bin\Debug\</OutputPath>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-  </PropertyGroup>
-  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
-    <DebugType>pdbonly</DebugType>
-    <Optimize>true</Optimize>
-    <OutputPath>bin\Release\</OutputPath>
-    <DefineConstants>TRACE</DefineConstants>
-    <ErrorReport>prompt</ErrorReport>
-    <WarningLevel>4</WarningLevel>
-  </PropertyGroup>
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|AnyCPU'">
-    <OutputPath>bin\Any CPU\</OutputPath>
-    <DefineConstants>TRACE</DefineConstants>
-    <Optimize>true</Optimize>
-    <DebugType>pdbonly</DebugType>
-    <PlatformTarget>AnyCPU</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x64'">
-    <DebugSymbols>true</DebugSymbols>
-    <OutputPath>bin\x64\Debug\</OutputPath>
-    <DefineConstants>DEBUG;TRACE</DefineConstants>
-    <DebugType>full</DebugType>
-    <PlatformTarget>x64</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Release|x64'">
-    <OutputPath>bin\x64\Release\</OutputPath>
-    <DefineConstants>TRACE</DefineConstants>
-    <Optimize>true</Optimize>
-    <DebugType>pdbonly</DebugType>
-    <PlatformTarget>x64</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|x64'">
-    <OutputPath>bin\x64\Any CPU\</OutputPath>
-    <DefineConstants>TRACE</DefineConstants>
-    <Optimize>true</Optimize>
-    <DebugType>pdbonly</DebugType>
-    <PlatformTarget>x64</PlatformTarget>
-    <ErrorReport>prompt</ErrorReport>
-    <CodeAnalysisRuleSet>MinimumRecommendedRules.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Debug|x86'">
-    <PlatformTarget>x86</PlatformTarget>
-    <OutputPath>bin\x86\Debug\</OutputPath>
-  </PropertyGroup>
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Release|x86'">
-    <PlatformTarget>x86</PlatformTarget>
-    <OutputPath>bin\x86\Release\</OutputPath>
-  </PropertyGroup>
-  <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'Any CPU|x86'">
-    <PlatformTarget>x86</PlatformTarget>
-    <OutputPath>bin\x86\Any CPU\</OutputPath>
   </PropertyGroup>
+
   <ItemGroup>
-    <Reference Include="Microsoft.VisualStudio.QualityTools.UnitTestFramework, Version=10.1.0.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL" />
+    <PackageReference Include="MSTest.TestAdapter" Version="1.1.18" />
-    <Reference Include="System" />
+    <PackageReference Include="MSTest.TestFramework" Version="1.1.18" />
-    <Reference Include="System.Core" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.3.0" />
   </ItemGroup>
+
   <ItemGroup>
-    <Compile Include="ByteReaderTests.cs" />
+    <ProjectReference Include="..\..\Frameworks\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj" />
-    <Compile Include="ByteWriterTests.cs" />
+    <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj" />
-    <Compile Include="ExtensionTests.cs" />
-    <Compile Include="MqttPacketSerializerTests.cs" />
-    <Compile Include="MqttServerTests.cs" />
-    <Compile Include="MqttSubscriptionsManagerTests.cs" />
-    <Compile Include="Properties\AssemblyInfo.cs" />
-    <Compile Include="TestMqttCommunicationAdapter.cs" />
-    <Compile Include="TestMqttServerAdapter.cs" />
-    <Compile Include="TopicFilterComparerTests.cs" />
   </ItemGroup>
-  <ItemGroup>
+
-    <Service Include="{82A7F48D-3B50-4B1E-B82E-3ADA8210C358}" />
-  </ItemGroup>
-  <ItemGroup>
-    <ProjectReference Include="..\..\MQTTnet.Core\MQTTnet.Core.csproj">
-      <Project>{2ecb99e4-72d0-4c23-99ba-93d511d3967d}</Project>
-      <Name>MQTTnet.Core</Name>
-    </ProjectReference>
-  </ItemGroup>
-  <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
-  <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
-       Other similar extension points exist, see Microsoft.Common.targets.
-  <Target Name="BeforeBuild">
-  </Target>
-  <Target Name="AfterBuild">
-  </Target>
-  -->
 </Project>

Tests/MQTTnet.Core.Tests/MqttCommunicationAdapterFactory.cs

@@ -0,0 +1,20 @@
+using MQTTnet.Core.Adapter;
+using MQTTnet.Core.Client;
+
+namespace MQTTnet.Core.Tests
+{
+    public class MqttCommunicationAdapterFactory : IMqttCommunicationAdapterFactory
+    {
+        private readonly IMqttCommunicationAdapter _adapter;
+
+        public MqttCommunicationAdapterFactory(IMqttCommunicationAdapter adapter)
+        {
+            _adapter = adapter;
+        }
+
+        public IMqttCommunicationAdapter CreateMqttCommunicationAdapter(MqttClientOptions options)
+        {
+            return _adapter;
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs

@@ -6,7 +6,6 @@ using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Channel;
-using MQTTnet.Core.Client;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;
 using MQTTnet.Core.Serializer;
@@ -411,7 +410,7 @@ namespace MQTTnet.Core.Tests
                 _stream.Position = 0;
             }
 
-            public Task ConnectAsync(MqttClientOptions options)
+            public Task ConnectAsync()
             {
                 return Task.FromResult(0);
             }

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -48,12 +48,13 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public async Task MqttServer_WillMessage()
         {
-            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() });
+            var serverAdapter = new TestMqttServerAdapter();
-            s.Start();
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
 
             var willMessage = new MqttApplicationMessage("My/last/will", new byte[0], MqttQualityOfServiceLevel.AtMostOnce, false);
-            var c1 = ConnectTestClient("c1", null, s);
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
-            var c2 = ConnectTestClient("c2", willMessage, s);
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2", willMessage);
 
             var receivedMessagesCount = 0;
             c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -63,7 +64,7 @@ namespace MQTTnet.Core.Tests
 
             await Task.Delay(1000);
 
-            s.Stop();
+            await s.StopAsync();
 
             Assert.AreEqual(1, receivedMessagesCount);
         }
@@ -71,15 +72,16 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public async Task MqttServer_Unsubscribe()
         {
-            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() });
+            var serverAdapter = new TestMqttServerAdapter();
-            s.Start();
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
 
-            var c1 = ConnectTestClient("c1", null, s);
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
-            var c2 = ConnectTestClient("c2", null, s);
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2");
 
             var receivedMessagesCount = 0;
             c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
-            
+
             var message = new MqttApplicationMessage("a", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, false);
 
             await c2.PublishAsync(message);
@@ -97,7 +99,7 @@ namespace MQTTnet.Core.Tests
             await Task.Delay(500);
             Assert.AreEqual(1, receivedMessagesCount);
 
-            s.Stop();
+            await s.StopAsync();
             await Task.Delay(500);
 
             Assert.AreEqual(1, receivedMessagesCount);
@@ -106,10 +108,11 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public async Task MqttServer_Publish()
         {
-            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() });
+            var serverAdapter = new TestMqttServerAdapter();
-            s.Start();
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
 
-            var c1 = ConnectTestClient("c1", null, s);
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
 
             var receivedMessagesCount = 0;
             c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -120,36 +123,140 @@ namespace MQTTnet.Core.Tests
             s.Publish(message);
             await Task.Delay(500);
 
-            s.Stop();
+            await s.StopAsync();
 
             Assert.AreEqual(1, receivedMessagesCount);
         }
 
-        private static MqttClient ConnectTestClient(string clientId, MqttApplicationMessage willMessage, MqttServer server)
+        [TestMethod]
+        public async Task MqttServer_NoRetainedMessage()
         {
-            var adapterA = new TestMqttCommunicationAdapter();
+            var serverAdapter = new TestMqttServerAdapter();
-            var adapterB = new TestMqttCommunicationAdapter();
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
-            adapterA.Partner = adapterB;
+            await s.StartAsync();
-            adapterB.Partner = adapterA;
 
-            var client = new MqttClient(new MqttClientOptions(), adapterA);
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
-            server.InjectClient(clientId, adapterB);
+            await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, false));
-            client.ConnectAsync(willMessage).Wait();
+            await c1.DisconnectAsync();
-            return client;
+
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2");
+            var receivedMessagesCount = 0;
+            c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
+            await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
+
+            await Task.Delay(500);
+
+            await s.StopAsync();
+
+            Assert.AreEqual(0, receivedMessagesCount);
         }
 
-        private async Task TestPublishAsync(
+        [TestMethod]
+        public async Task MqttServer_RetainedMessage()
+        {
+            var serverAdapter = new TestMqttServerAdapter();
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
+
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
+            await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true));
+            await c1.DisconnectAsync();
+
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2");
+            var receivedMessagesCount = 0;
+            c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
+            await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
+
+            await Task.Delay(500);
+
+            await s.StopAsync();
+
+            Assert.AreEqual(1, receivedMessagesCount);
+        }
+
+        [TestMethod]
+        public async Task MqttServer_ClearRetainedMessage()
+        {
+            var serverAdapter = new TestMqttServerAdapter();
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
+
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
+            await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true));
+            await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[0], MqttQualityOfServiceLevel.AtLeastOnce, true));
+            await c1.DisconnectAsync();
+
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2");
+            var receivedMessagesCount = 0;
+            c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
+            await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
+
+            await Task.Delay(500);
+
+            await s.StopAsync();
+
+            Assert.AreEqual(0, receivedMessagesCount);
+        }
+
+        [TestMethod]
+        public async Task MqttServer_PersistRetainedMessage()
+        {
+            var storage = new TestStorage();
+
+            var serverAdapter = new TestMqttServerAdapter();
+            var s = new MqttServer(new MqttServerOptions { Storage = storage }, new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
+
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
+            await c1.PublishAsync(new MqttApplicationMessage("retained", new byte[3], MqttQualityOfServiceLevel.AtLeastOnce, true));
+            await c1.DisconnectAsync();
+
+            await s.StopAsync();
+
+            s = new MqttServer(new MqttServerOptions { Storage = storage }, new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
+
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2");
+            var receivedMessagesCount = 0;
+            c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
+            await c2.SubscribeAsync(new TopicFilter("retained", MqttQualityOfServiceLevel.AtLeastOnce));
+
+            await Task.Delay(500);
+
+            await s.StopAsync();
+
+            Assert.AreEqual(1, receivedMessagesCount);
+        }
+
+        private class TestStorage : IMqttServerStorage
+        {
+            private IList<MqttApplicationMessage> _messages = new List<MqttApplicationMessage>();
+
+            public Task SaveRetainedMessagesAsync(IList<MqttApplicationMessage> messages)
+            {
+                _messages = messages;
+                return Task.FromResult(0);
+            }
+
+            public Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync()
+            {
+                return Task.FromResult(_messages);
+            }
+        }
+
+        private static async Task TestPublishAsync(
             string topic,
             MqttQualityOfServiceLevel qualityOfServiceLevel,
             string topicFilter,
             MqttQualityOfServiceLevel filterQualityOfServiceLevel,
             int expectedReceivedMessagesCount)
         {
-            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { new TestMqttServerAdapter() });
+            var serverAdapter = new TestMqttServerAdapter();
-            s.Start();
+            var s = new MqttServer(new MqttServerOptions(), new List<IMqttServerAdapter> { serverAdapter });
+            await s.StartAsync();
 
-            var c1 = ConnectTestClient("c1", null, s);
+            var c1 = await serverAdapter.ConnectTestClient(s, "c1");
-            var c2 = ConnectTestClient("c2", null, s);
+            var c2 = await serverAdapter.ConnectTestClient(s, "c2");
 
             var receivedMessagesCount = 0;
             c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
@@ -162,7 +269,7 @@ namespace MQTTnet.Core.Tests
 
             await Task.Delay(500);
 
-            s.Stop();
+            await s.StopAsync();
 
             Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount);
         }

Tests/MQTTnet.Core.Tests/Properties/AssemblyInfo.cs

@@ -1,15 +0,0 @@
-using System.Reflection;
-using System.Runtime.InteropServices;
-
-[assembly: AssemblyTitle("MQTTnet.Core.Tests")]
-[assembly: AssemblyDescription("")]
-[assembly: AssemblyConfiguration("")]
-[assembly: AssemblyCompany("Christian Kratky")]
-[assembly: AssemblyProduct("MQTTnet")]
-[assembly: AssemblyCopyright("Copyright © Christian Kratky 2016-2017")]
-[assembly: AssemblyTrademark("")]
-[assembly: AssemblyCulture("")]
-[assembly: ComVisible(false)]
-[assembly: Guid("a7ff0c91-25de-4ba6-b39e-f54e8dadf1cc")]
-[assembly: AssemblyVersion("1.0.0.0")]
-[assembly: AssemblyFileVersion("1.0.0.0")]

Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs

@@ -4,7 +4,6 @@ using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
-using MQTTnet.Core.Client;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Serializer;
 
@@ -18,7 +17,7 @@ namespace MQTTnet.Core.Tests
 
         public IMqttPacketSerializer PacketSerializer { get; } = new MqttPacketSerializer();
 
-        public Task ConnectAsync(TimeSpan timeout, MqttClientOptions options)
+        public Task ConnectAsync(TimeSpan timeout)
         {
             return Task.FromResult(0);
         }

Tests/MQTTnet.Core.Tests/TestMqttServerAdapter.cs

@@ -1,24 +1,63 @@
 using System;
+using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Server;
+using MQTTnet.Core.Client;
 
 namespace MQTTnet.Core.Tests
 {
     public class TestMqttServerAdapter : IMqttServerAdapter
     {
-        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+        public event Action<IMqttCommunicationAdapter> ClientAccepted;
 
-        public void FireClientConnectedEvent(MqttClientConnectedEventArgs eventArgs)
+        public async Task<MqttClient> ConnectTestClient(IMqttServer server, string clientId, MqttApplicationMessage willMessage = null)
         {
-            ClientConnected?.Invoke(this, eventArgs);
+            var adapterA = new TestMqttCommunicationAdapter();
+            var adapterB = new TestMqttCommunicationAdapter();
+            adapterA.Partner = adapterB;
+            adapterB.Partner = adapterA;
+
+            var client = new MqttClient(new MqttCommunicationAdapterFactory(adapterA));
+            var connected = WaitForClientToConnect(server, clientId);
+
+            FireClientAcceptedEvent(adapterB);
+            await client.ConnectAsync(new MqttClientTcpOptions { ClientId = clientId, WillMessage = willMessage });
+            await connected;
+
+            return client;
+        }
+        
+        private static Task WaitForClientToConnect(IMqttServer s, string clientId)
+        {
+            var tcs = new TaskCompletionSource<object>();
+
+            void Handler(object sender, MqttClientConnectedEventArgs args)
+            {
+                if (args.Client.ClientId == clientId)
+                {
+                    s.ClientConnected -= Handler;
+                    tcs.SetResult(null);
+                }
+            }
+
+            s.ClientConnected += Handler;
+
+            return tcs.Task;
         }
 
-        public void Start(MqttServerOptions options)
+        private void FireClientAcceptedEvent(IMqttCommunicationAdapter adapter)
         {
+            ClientAccepted?.Invoke(adapter);
         }
 
-        public void Stop()
+        public Task StartAsync(MqttServerOptions options)
-        {                 
+        {
+            return Task.FromResult(0);
+        }
+
+        public Task StopAsync()
+        {
+            return Task.FromResult(0);
         }
     }
 }

Tests/MQTTnet.TestApp.NetCore/Program.cs

@@ -35,7 +35,7 @@ namespace MQTTnet.TestApp.NetCore
         private static async Task RunClientAsync(string[] arguments)
         {
 
-            MqttTrace.TraceMessagePublished += (s, e) =>
+            MqttNetTrace.TraceMessagePublished += (s, e) =>
             {
                 Console.WriteLine($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}");
                 if (e.Exception != null)
@@ -46,15 +46,14 @@ namespace MQTTnet.TestApp.NetCore
 
             try
             {
-                var options = new MqttClientOptions
+                var options = new MqttClientWebSocketOptions
                 {
-                    Server = "localhost",
+                    Uri = "localhost",
                     ClientId = "XYZ",
-                    CleanSession = true,
+                    CleanSession = true
-                    ConnectionType = MqttConnectionType.Ws 
                 };
 
-                var client = new MqttClientFactory().CreateMqttClient(options);
+                var client = new MqttClientFactory().CreateMqttClient();
                 client.ApplicationMessageReceived += (s, e) =>
                 {
                     Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
@@ -84,7 +83,7 @@ namespace MQTTnet.TestApp.NetCore
 
                     try
                     {
-                        await client.ConnectAsync();
+                        await client.ConnectAsync(options);
                     }
                     catch
                     {
@@ -94,7 +93,7 @@ namespace MQTTnet.TestApp.NetCore
 
                 try
                 {
-                    await client.ConnectAsync();
+                    await client.ConnectAsync(options);
                 }
                 catch (Exception exception)
                 {
@@ -125,7 +124,7 @@ namespace MQTTnet.TestApp.NetCore
 
         private static void RunServerAsync(string[] arguments)
         {
-            MqttTrace.TraceMessagePublished += (s, e) =>
+            MqttNetTrace.TraceMessagePublished += (s, e) =>
             {
                 Console.WriteLine($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}");
                 if (e.Exception != null)
@@ -153,12 +152,12 @@ namespace MQTTnet.TestApp.NetCore
                 };
 
                 var mqttServer = new MqttServerFactory().CreateMqttServer(options);
-                mqttServer.Start();
+                mqttServer.StartAsync();
 
                 Console.WriteLine("Press any key to exit.");
                 Console.ReadLine();
 
-                mqttServer.Stop();
+                mqttServer.StopAsync();
             }
             catch (Exception e)
             {

Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs

@@ -32,7 +32,7 @@ namespace MQTTnet.TestApp.NetFramework
         {
             try
             {
-                var options = new MqttClientOptions
+                var options = new MqttClientTcpOptions
                 {
                     Server = "localhost",
                     ClientId = "XYZ",
@@ -40,7 +40,7 @@ namespace MQTTnet.TestApp.NetFramework
                     DefaultCommunicationTimeout = TimeSpan.FromMinutes(10)
                 };
 
-                var client = new MqttClientFactory().CreateMqttClient(options);
+                var client = new MqttClientFactory().CreateMqttClient();
                 client.ApplicationMessageReceived += (s, e) =>
                 {
                 };
@@ -64,7 +64,7 @@ namespace MQTTnet.TestApp.NetFramework
 
                     try
                     {
-                        await client.ConnectAsync();
+                        await client.ConnectAsync(options);
                     }
                     catch
                     {
@@ -74,7 +74,7 @@ namespace MQTTnet.TestApp.NetFramework
 
                 try
                 {
-                    await client.ConnectAsync();
+                    await client.ConnectAsync(options);
                 }
                 catch (Exception exception)
                 {
@@ -167,7 +167,7 @@ namespace MQTTnet.TestApp.NetFramework
            });
         }
 
-        private static void RunServerAsync()
+        private static async Task RunServerAsync()
         {
             try
             {
@@ -201,12 +201,12 @@ namespace MQTTnet.TestApp.NetFramework
                         stopwatch.Restart();
                     }
                 };
-                mqttServer.Start();
+                await mqttServer.StartAsync();
 
                 Console.WriteLine("Press any key to exit.");
                 Console.ReadLine();
 
-                mqttServer.Stop();
+                await mqttServer.StopAsync();
             }
             catch (Exception e)
             {

Tests/MQTTnet.TestApp.NetFramework/Program.cs

@@ -9,6 +9,7 @@ using MQTTnet.Core.Diagnostics;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;
 using MQTTnet.Core.Server;
+using MQTTnet.Implementations;
 
 namespace MQTTnet.TestApp.NetFramework
 {
@@ -41,7 +42,7 @@ namespace MQTTnet.TestApp.NetFramework
         private static async Task RunClientAsync(string[] arguments)
         {
 
-            MqttTrace.TraceMessagePublished += (s, e) =>
+            MqttNetTrace.TraceMessagePublished += (s, e) =>
             {
                 Console.WriteLine($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}");
                 if (e.Exception != null)
@@ -52,15 +53,20 @@ namespace MQTTnet.TestApp.NetFramework
 
             try
             {
-                var options = new MqttClientOptions
+                var options = new MqttClientWebSocketOptions
                 {
-                    Server = "localhost",
+                    Uri = "broker.hivemq.com:8000/mqtt"
-                    ClientId = "XYZ",
-                    CleanSession = true
                 };
 
-                var client = new MqttClientFactory().CreateMqttClient(options);
+                ////var options = new MqttClientOptions
-                client.ApplicationMessageReceived += (s, e) =>
+                ////{
+                ////    Server = "localhost",
+                ////    ClientId = "XYZ",
+                ////    CleanSession = true
+                ////};
+
+                var mqttClient = new MqttClientFactory().CreateMqttClient();
+                mqttClient.ApplicationMessageReceived += (s, e) =>
                 {
                     Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
                     Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
@@ -70,24 +76,24 @@ namespace MQTTnet.TestApp.NetFramework
                     Console.WriteLine();
                 };
 
-                client.Connected += async (s, e) =>
+                mqttClient.Connected += async (s, e) =>
                 {
                     Console.WriteLine("### CONNECTED WITH SERVER ###");
 
-                    await client.SubscribeAsync(new List<TopicFilter>
+                    await mqttClient.SubscribeAsync(new List<TopicFilter>
                     {
                         new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)
                     });
                 };
 
-                client.Disconnected += async (s, e) =>
+                mqttClient.Disconnected += async (s, e) =>
                 {
                     Console.WriteLine("### DISCONNECTED FROM SERVER ###");
                     await Task.Delay(TimeSpan.FromSeconds(5));
 
                     try
                     {
-                        await client.ConnectAsync();
+                        await mqttClient.ConnectAsync(options);
                     }
                     catch
                     {
@@ -97,7 +103,7 @@ namespace MQTTnet.TestApp.NetFramework
 
                 try
                 {
-                    await client.ConnectAsync();
+                    await mqttClient.ConnectAsync(options);
                 }
                 catch (Exception exception)
                 {
@@ -112,7 +118,7 @@ namespace MQTTnet.TestApp.NetFramework
                     Console.ReadLine();
 
                     var applicationMessage = messageFactory.CreateApplicationMessage("myTopic", "Hello World", MqttQualityOfServiceLevel.AtLeastOnce);
-                    await client.PublishAsync(applicationMessage);
+                    await mqttClient.PublishAsync(applicationMessage);
                 }
             }
             catch (Exception exception)
@@ -121,9 +127,9 @@ namespace MQTTnet.TestApp.NetFramework
             }
         }
 
-        private static void RunServerAsync(string[] arguments)
+        private static async Task RunServerAsync(string[] arguments)
         {
-            MqttTrace.TraceMessagePublished += (s, e) =>
+            MqttNetTrace.TraceMessagePublished += (s, e) =>
             {
                 Console.WriteLine($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}");
                 if (e.Exception != null)
@@ -151,12 +157,12 @@ namespace MQTTnet.TestApp.NetFramework
                 };
 
                 var mqttServer = new MqttServerFactory().CreateMqttServer(options);
-                mqttServer.Start();
+                await mqttServer.StartAsync();
 
                 Console.WriteLine("Press any key to exit.");
                 Console.ReadLine();
 
-                mqttServer.Stop();
+                await mqttServer.StopAsync();
             }
             catch (Exception e)
             {
@@ -165,5 +171,19 @@ namespace MQTTnet.TestApp.NetFramework
 
             Console.ReadLine();
         }
+
+        private static async Task WikiCode()
+        {
+            // For .NET Framwork & netstandard apps:
+            MqttTcpChannel.CustomCertificateValidationCallback = (x509Certificate, x509Chain, sslPolicyErrors, mqttClientTcpOptions) =>
+            {
+                if (mqttClientTcpOptions.Server == "server_with_revoked_cert")
+                {
+                    return true;
+                }
+
+                return false;
+            };
+        }
     }
-}
+}

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

@@ -1,5 +1,7 @@
 using System;
 using System.Text;
+using System.Threading.Tasks;
+using Windows.Security.Cryptography.Certificates;
 using Windows.UI.Core;
 using Windows.UI.Xaml;
 using MQTTnet.Core;
@@ -7,6 +9,8 @@ using MQTTnet.Core.Client;
 using MQTTnet.Core.Diagnostics;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;
+using MQTTnet.Core.Server;
+using MQTTnet.Implementations;
 
 namespace MQTTnet.TestApp.UniversalWindows
 {
@@ -18,10 +22,10 @@ namespace MQTTnet.TestApp.UniversalWindows
         {
             InitializeComponent();
 
-            MqttTrace.TraceMessagePublished += OnTraceMessagePublished;
+            MqttNetTrace.TraceMessagePublished += OnTraceMessagePublished;
         }
 
-        private async void OnTraceMessagePublished(object sender, MqttTraceMessagePublishedEventArgs e)
+        private async void OnTraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e)
         {
             await Trace.Dispatcher.RunAsync(CoreDispatcherPriority.High, () =>
             {
@@ -37,16 +41,36 @@ namespace MQTTnet.TestApp.UniversalWindows
 
         private async void Connect(object sender, RoutedEventArgs e)
         {
-            var options = new MqttClientOptions
+            MqttClientOptions options = null;
+            if (UseTcp.IsChecked == true)
             {
-                Server = Server.Text,
+                options = new MqttClientTcpOptions
-                UserName = User.Text,
+                {
-                Password = Password.Text,
+                    Server = Server.Text
-                ClientId = ClientId.Text,
+                };
-                TlsOptions = { UseTls = UseTls.IsChecked == true },
+            }
-                ConnectionType = UseTcp.IsChecked == true ? MqttConnectionType.Tcp : MqttConnectionType.Ws
+
-            };
+            if (UseWs.IsChecked == true)
-            
+            {
+                options = new MqttClientWebSocketOptions
+                {
+                    Uri = Server.Text
+                };
+            }
+
+            if (options == null)
+            {
+                throw new InvalidOperationException();
+            }
+
+            options.UserName = User.Text;
+            options.Password = Password.Text;
+            options.ClientId = ClientId.Text;
+            options.TlsOptions.UseTls = UseTls.IsChecked == true;
+            options.TlsOptions.IgnoreCertificateChainErrors = true;
+            options.TlsOptions.IgnoreCertificateRevocationErrors = true;
+            options.TlsOptions.AllowUntrustedCertificates = true;
+
             try
             {
                 if (_mqttClient != null)
@@ -55,8 +79,8 @@ namespace MQTTnet.TestApp.UniversalWindows
                 }
 
                 var factory = new MqttClientFactory();
-                _mqttClient = factory.CreateMqttClient(options);
+                _mqttClient = factory.CreateMqttClient();
-                await _mqttClient.ConnectAsync();
+                await _mqttClient.ConnectAsync(options);
             }
             catch (Exception exception)
             {
@@ -170,5 +194,99 @@ namespace MQTTnet.TestApp.UniversalWindows
                 Trace.Text += exception + Environment.NewLine;
             }
         }
+
+        private async Task WikiCode()
+        {
+            var mqttClient = new MqttClientFactory().CreateMqttClient();
+
+            // ----------------------------------
+
+            var tcpOptions = new MqttClientTcpOptions
+            {
+                Server = "broker.hivemq.org",
+                ClientId = "TestClient"
+            };
+
+            await mqttClient.ConnectAsync(tcpOptions);
+
+            // ----------------------------------
+
+            var secureTcpOptions = new MqttClientTcpOptions
+            {
+                Server = "broker.hivemq.org",
+                ClientId = "TestClient",
+                TlsOptions = new MqttClientTlsOptions
+                {
+                    UseTls = true,
+                    IgnoreCertificateChainErrors = true,
+                    IgnoreCertificateRevocationErrors = true,
+                    AllowUntrustedCertificates = true
+                }
+            };
+
+            // ----------------------------------
+
+            var wsOptions = new MqttClientWebSocketOptions
+            {
+                Uri = "broker.hivemq.com:8000/mqtt",
+                ClientId = "TestClient"
+            };
+
+            await mqttClient.ConnectAsync(wsOptions);
+
+            // ----------------------------------
+            {
+                var options = new MqttServerOptions();
+
+                var mqttServer = new MqttServerFactory().CreateMqttServer(options);
+                await mqttServer.StartAsync();
+
+                Console.WriteLine("Press any key to exit.");
+                Console.ReadLine();
+
+                await mqttServer.StopAsync();
+            }
+
+            // ----------------------------------
+
+            {
+                var options = new MqttServerOptions
+                {
+                    ConnectionValidator = c =>
+                    {
+                        if (c.ClientId.Length < 10)
+                        {
+                            return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
+                        }
+
+                        if (c.Username != "mySecretUser")
+                        {
+                            return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
+                        }
+
+                        if (c.Password != "mySecretPassword")
+                        {
+                            return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
+                        }
+
+                        return MqttConnectReturnCode.ConnectionAccepted;
+                    }
+                };
+            }
+
+            // ----------------------------------
+
+            // For UWP apps:
+            MqttTcpChannel.CustomIgnorableServerCertificateErrorsResolver = o =>
+            {
+                if (o.Server == "server_with_revoked_cert")
+                {
+                    return new[] { ChainValidationResult.Revoked };
+                }
+
+                return new ChainValidationResult[0];
+            };
+
+        }
     }
 }