Merge branch 'master' into feature-isconnected

Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs

@@ -1,17 +1,16 @@
-using System;
-using System.Net;
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.AspNetCore.Client.Tcp;
-using MQTTnet.Client;
 using MQTTnet.Client.Options;
 using MQTTnet.Diagnostics;
 using MQTTnet.Formatter;
+using System;
+using System.Net;
 
 namespace MQTTnet.AspNetCore.Client
 {
     public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
     {
-        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 

Source/MQTTnet.AspnetCore/Extensions/ServiceCollectionExtensions.cs

@@ -1,10 +1,10 @@
-using System;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
-using MQTTnet.Server;
 using MQTTnet.Implementations;
-using Microsoft.Extensions.DependencyInjection;
+using MQTTnet.Server;
+using System;
 
 namespace MQTTnet.AspNetCore
 {
@@ -13,7 +13,7 @@ namespace MQTTnet.AspNetCore
         public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, IMqttServerOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
-            
+
             services.AddSingleton(options);
 
             services.AddHostedMqttServer();
@@ -23,7 +23,8 @@ namespace MQTTnet.AspNetCore
 
         public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure)
         {
-            services.AddSingleton<IMqttServerOptions>(s => {
+            services.AddSingleton<IMqttServerOptions>(s =>
+            {
                 var builder = new MqttServerOptionsBuilder();
                 configure(builder);
                 return builder.Build();
@@ -36,7 +37,8 @@ namespace MQTTnet.AspNetCore
 
         public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)
         {
-            services.AddSingleton<IMqttServerOptions>(s => {
+            services.AddSingleton<IMqttServerOptions>(s =>
+            {
                 var builder = new AspNetMqttServerOptionsBuilder(s);
                 configure(builder);
                 return builder.Build();
@@ -60,14 +62,12 @@ namespace MQTTnet.AspNetCore
         private static IServiceCollection AddHostedMqttServer(this IServiceCollection services)
         {
             var logger = new MqttNetLogger();
-            var childLogger = logger.CreateChildLogger();
 
             services.AddSingleton<IMqttNetLogger>(logger);
-            services.AddSingleton(childLogger);
             services.AddSingleton<MqttHostedServer>();
             services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
             services.AddSingleton<IMqttServer>(s => s.GetService<MqttHostedServer>());
-            
+
             return services;
         }
 

Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj

@@ -21,10 +21,10 @@
   </ItemGroup>
     
   <ItemGroup Condition="'$(TargetFramework)' != 'netcoreapp3.1'">
-    <PackageReference Include="Microsoft.AspNetCore.Connections.Abstractions" Version="2.1.3" />
-    <PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.0.3" />
-    <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.1.1" />
-    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.1.1" />
+    <PackageReference Include="Microsoft.AspNetCore.Connections.Abstractions" Version="3.1.3" />
+    <PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.1.0" />
+    <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.2.1" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.3" />
   </ItemGroup>
 
   <ItemGroup>

Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs

@@ -1,20 +1,20 @@
-using System;
-using System.Net.WebSockets;
-using System.Threading.Tasks;
-using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics;
 using MQTTnet.Formatter;
 using MQTTnet.Implementations;
 using MQTTnet.Server;
+using System;
+using System.Net.WebSockets;
+using System.Threading.Tasks;
 
 namespace MQTTnet.AspNetCore
 {
     public class MqttWebSocketServerAdapter : IMqttServerAdapter
     {
-        private readonly IMqttNetChildLogger _logger;
+        private readonly IMqttNetLogger _logger;
 
-        public MqttWebSocketServerAdapter(IMqttNetChildLogger logger)
+        public MqttWebSocketServerAdapter(IMqttNetLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
@@ -38,7 +38,7 @@ namespace MQTTnet.AspNetCore
             if (webSocket == null) throw new ArgumentNullException(nameof(webSocket));
 
             var endpoint = $"{httpContext.Connection.RemoteIpAddress}:{httpContext.Connection.RemotePort}";
-            
+
             var clientCertificate = await httpContext.Connection.GetClientCertificateAsync().ConfigureAwait(false);
             try
             {

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -1,9 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Client;
+using MQTTnet.Client;
 using MQTTnet.Client.Connecting;
 using MQTTnet.Client.Disconnecting;
 using MQTTnet.Client.Publishing;
@@ -13,6 +8,11 @@ using MQTTnet.Exceptions;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Extensions.ManagedClient
 {
@@ -33,7 +33,7 @@ namespace MQTTnet.Extensions.ManagedClient
         private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
 
         private readonly IMqttClient _mqttClient;
-        private readonly IMqttNetChildLogger _logger;
+        private readonly IMqttNetLogger _logger;
 
         private readonly AsyncLock _messageQueueLock = new AsyncLock();
 
@@ -42,8 +42,8 @@ namespace MQTTnet.Extensions.ManagedClient
         private Task _maintainConnectionTask;
 
         private ManagedMqttClientStorageManager _storageManager;
-        
-        public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
+
+        public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
         {
             _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
 

Source/MQTTnet.Extensions.ManagedClient/MqttFactoryExtensions.cs

@@ -1,5 +1,5 @@
-using System;
-using MQTTnet.Diagnostics;
+using MQTTnet.Diagnostics;
+using System;
 
 namespace MQTTnet.Extensions.ManagedClient
 {
@@ -9,7 +9,7 @@ namespace MQTTnet.Extensions.ManagedClient
         {
             if (factory == null) throw new ArgumentNullException(nameof(factory));
 
-            return new ManagedMqttClient(factory.CreateMqttClient(), factory.DefaultLogger.CreateChildLogger());
+            return new ManagedMqttClient(factory.CreateMqttClient(), factory.DefaultLogger);
         }
 
         public static IManagedMqttClient CreateManagedMqttClient(this IMqttFactory factory, IMqttNetLogger logger)
@@ -17,7 +17,7 @@ namespace MQTTnet.Extensions.ManagedClient
             if (factory == null) throw new ArgumentNullException(nameof(factory));
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
-            return new ManagedMqttClient(factory.CreateMqttClient(logger), logger.CreateChildLogger());
+            return new ManagedMqttClient(factory.CreateMqttClient(logger), logger);
         }
     }
 }

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs

@@ -1,15 +1,15 @@
-using System;
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.Client.Options;
 using MQTTnet.Diagnostics;
 using MQTTnet.Formatter;
 using MQTTnet.Implementations;
+using System;
 
 namespace MQTTnet.Extensions.WebSocket4Net
 {
     public class WebSocket4NetMqttClientAdapterFactory : IMqttClientAdapterFactory
     {
-        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
             if (logger == null) throw new ArgumentNullException(nameof(logger));
@@ -17,19 +17,19 @@ namespace MQTTnet.Extensions.WebSocket4Net
             switch (options.ChannelOptions)
             {
                 case MqttClientTcpOptions _:
-                {
-                    return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
-                }
+                    {
+                        return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
+                    }
 
                 case MqttClientWebSocketOptions webSocketOptions:
-                {
-                    return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
-                }
+                    {
+                        return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
+                    }
 
                 default:
-                {
-                    throw new NotSupportedException();
-                }
+                    {
+                        throw new NotSupportedException();
+                    }
             }
         }
     }

Source/MQTTnet.Server/Logging/MqttNetChildLoggerWrapper.cs

@@ -1,43 +1,43 @@
-using System;
-using MQTTnet.Diagnostics;
+//using MQTTnet.Diagnostics;
+//using System;
 
-namespace MQTTnet.Server.Logging
-{
-    public class MqttNetChildLoggerWrapper : IMqttNetChildLogger
-    {
-        private readonly MqttNetLoggerWrapper _logger;
-        private readonly string _source;
+//namespace MQTTnet.Server.Logging
+//{
+//    public class MqttNetChildLoggerWrapper : IMqttNetChildLogger
+//    {
+//        private readonly MqttNetLoggerWrapper _logger;
+//        private readonly string _source;
 
-        public MqttNetChildLoggerWrapper(string source, MqttNetLoggerWrapper logger)
-        {
-            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+//        public MqttNetChildLoggerWrapper(string source, MqttNetLoggerWrapper logger)
+//        {
+//            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
-            _source = source;
-        }
+//            _source = source;
+//        }
 
-        public IMqttNetChildLogger CreateChildLogger(string source = null)
-        {
-            return _logger.CreateChildLogger(source);
-        }
+//        public IMqttNetLogger CreateChildLogger(string source = null)
+//        {
+//            return _logger.CreateChildLogger(source);
+//        }
 
-        public void Verbose(string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null);
-        }
+//        public void Verbose(string message, params object[] parameters)
+//        {
+//            _logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null);
+//        }
 
-        public void Info(string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null);
-        }
+//        public void Info(string message, params object[] parameters)
+//        {
+//            _logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null);
+//        }
 
-        public void Warning(Exception exception, string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception);
-        }
+//        public void Warning(Exception exception, string message, params object[] parameters)
+//        {
+//            _logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception);
+//        }
 
-        public void Error(Exception exception, string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception);
-        }
-    }
-}
+//        public void Error(Exception exception, string message, params object[] parameters)
+//        {
+//            _logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception);
+//        }
+//    }
+//}

Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs

@@ -1,13 +1,13 @@
-using System;
-using System.Threading;
-using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging;
 using MQTTnet.Diagnostics;
+using System;
+using System.Threading;
 
 namespace MQTTnet.Server.Logging
 {
     public class MqttNetLoggerWrapper : IMqttNetLogger
     {
-        private readonly ILogger<MqttServer> _logger;
+        readonly ILogger<MqttServer> _logger;
 
         public MqttNetLoggerWrapper(ILogger<MqttServer> logger)
         {
@@ -16,25 +16,39 @@ namespace MQTTnet.Server.Logging
 
         public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
 
-        public IMqttNetChildLogger CreateChildLogger(string source = null)
+        public IMqttNetLogger CreateChildLogger(string source)
         {
-            return new MqttNetChildLoggerWrapper(source, this);
+            return new MqttNetLogger(source);
         }
 
-        public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
+        public void Publish(MqttNetLogLevel level, string source, string message, object[] parameters, Exception exception)
         {
-            var convertedLogLevel = ConvertLogLevel(logLevel);
+            var convertedLogLevel = ConvertLogLevel(level);
             _logger.Log(convertedLogLevel, exception, message, parameters);
 
             var logMessagePublishedEvent = LogMessagePublished;
             if (logMessagePublishedEvent != null)
             {
-                var logMessage = new MqttNetLogMessage(null, DateTime.UtcNow, Thread.CurrentThread.ManagedThreadId, source, logLevel, message, exception);
+                var logMessage = new MqttNetLogMessage
+                {
+                    Timestamp = DateTime.UtcNow,
+                    ThreadId = Thread.CurrentThread.ManagedThreadId,
+                    Source = source,
+                    Level = level,
+                    Message = message,
+                    Exception = exception
+                };
+
                 logMessagePublishedEvent.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage));
             }
         }
-        
-        private static LogLevel ConvertLogLevel(MqttNetLogLevel logLevel)
+
+        public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception)
+        {
+            Publish(logLevel, null, message, parameters, exception);
+        }
+
+        static LogLevel ConvertLogLevel(MqttNetLogLevel logLevel)
         {
             switch (logLevel)
             {

Source/MQTTnet.Server/MQTTnet.Server.csproj

@@ -45,8 +45,10 @@
   <ItemGroup>
     <PackageReference Include="IronPython" Version="2.7.9" />
     <PackageReference Include="IronPython.StdLib" Version="2.7.9" />
-    <PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="5.0.0-rc2" />
-    <PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="5.0.0-rc2" />
+    <PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="3.1.3" />
+    <PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
+    <PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="5.3.1" />
+    <PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="5.3.1" />
   </ItemGroup>
 
   <ItemGroup>

Source/MQTTnet.Server/Mqtt/MqttServerService.cs

@@ -1,11 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Net.WebSockets;
-using System.Security.Authentication;
-using System.Text;
-using System.Threading.Tasks;
-using IronPython.Runtime;
+using IronPython.Runtime;
 using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.Logging;
 using MQTTnet.Adapter;
@@ -16,6 +9,13 @@ using MQTTnet.Protocol;
 using MQTTnet.Server.Configuration;
 using MQTTnet.Server.Scripting;
 using MQTTnet.Server.Status;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net.WebSockets;
+using System.Security.Authentication;
+using System.Text;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Server.Mqtt
 {
@@ -65,11 +65,11 @@ namespace MQTTnet.Server.Mqtt
             _pythonScriptHostService = pythonScriptHostService ?? throw new ArgumentNullException(nameof(pythonScriptHostService));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
-            _webSocketServerAdapter = new MqttWebSocketServerAdapter(mqttFactory.Logger.CreateChildLogger());
+            _webSocketServerAdapter = new MqttWebSocketServerAdapter(mqttFactory.Logger);
 
             var adapters = new List<IMqttServerAdapter>
             {
-                new MqttTcpServerAdapter(mqttFactory.Logger.CreateChildLogger())
+                new MqttTcpServerAdapter(mqttFactory.Logger)
                 {
                     TreatSocketOpeningErrorAsWarning = true // Opening other ports than for HTTP is not allows in Azure App Services.
                 },
@@ -215,7 +215,7 @@ namespace MQTTnet.Server.Mqtt
                 options
                     .WithEncryptedEndpoint()
                     .WithEncryptionSslProtocol(SslProtocols.Tls12);
-                
+
                 if (!string.IsNullOrEmpty(_settings.EncryptedTcpEndPoint?.Certificate?.Path))
                 {
                     IMqttServerCertificateCredentials certificateCredentials = null;
@@ -230,7 +230,7 @@ namespace MQTTnet.Server.Mqtt
 
                     options.WithEncryptionCertificate(_settings.EncryptedTcpEndPoint.Certificate.ReadCertificate(), certificateCredentials);
                 }
-                
+
                 if (_settings.EncryptedTcpEndPoint.TryReadIPv4(out var address4))
                 {
                     options.WithEncryptedEndpointBoundIPAddress(address4);

Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs

@@ -5,6 +5,6 @@ namespace MQTTnet.Adapter
 {
     public interface IMqttClientAdapterFactory
     {
-        IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger);
+        IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger);
     }
 }

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -14,12 +14,12 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.Adapter
 {
-    public class MqttChannelAdapter : Disposable, IMqttChannelAdapter
+    public sealed class MqttChannelAdapter : Disposable, IMqttChannelAdapter
     {
         const uint ErrorOperationAborted = 0x800703E3;
         const int ReadBufferSize = 4096;  // TODO: Move buffer size to config
 
-        readonly IMqttNetChildLogger _logger;
+        readonly IMqttNetLogger _logger;
         readonly IMqttChannel _channel;
         readonly MqttPacketReader _packetReader;
 
@@ -30,7 +30,7 @@ namespace MQTTnet.Adapter
         long _bytesReceived;
         long _bytesSent;
 
-        public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetChildLogger logger)
+        public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 

Source/MQTTnet/Client/MqttClient.cs

@@ -1,7 +1,3 @@
-using System;
-using System.Diagnostics;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Client.Connecting;
 using MQTTnet.Client.Disconnecting;
@@ -17,6 +13,10 @@ using MQTTnet.Internal;
 using MQTTnet.PacketDispatcher;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Client
 {
@@ -29,7 +29,7 @@ namespace MQTTnet.Client
         private readonly object _disconnectLock = new object();
 
         private readonly IMqttClientAdapterFactory _adapterFactory;
-        private readonly IMqttNetChildLogger _logger;
+        private readonly IMqttNetLogger _logger;
 
         private CancellationTokenSource _backgroundCancellationTokenSource;
         private Task _packetReceiverTask;

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -1,19 +1,19 @@
-using System;
+using MQTTnet.Client.ExtendedAuthenticationExchange;
+using MQTTnet.Formatter;
+using System;
 using System.Linq;
 using System.Text;
-using MQTTnet.Client.ExtendedAuthenticationExchange;
-using MQTTnet.Formatter;
 
 namespace MQTTnet.Client.Options
 {
     public class MqttClientOptionsBuilder
     {
-        private readonly MqttClientOptions _options = new MqttClientOptions();
+        readonly MqttClientOptions _options = new MqttClientOptions();
 
-        private MqttClientTcpOptions _tcpOptions;
-        private MqttClientWebSocketOptions _webSocketOptions;
-        private MqttClientOptionsBuilderTlsParameters _tlsParameters;
-        private MqttClientWebSocketProxyOptions _proxyOptions;
+        MqttClientTcpOptions _tcpOptions;
+        MqttClientWebSocketOptions _webSocketOptions;
+        MqttClientOptionsBuilderTlsParameters _tlsParameters;
+        MqttClientWebSocketProxyOptions _proxyOptions;
 
         public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
         {

Source/MQTTnet/Diagnostics/IMqttNetChildLogger.cs

@@ -1,17 +0,0 @@
-using System;
-
-namespace MQTTnet.Diagnostics
-{
-    public interface IMqttNetChildLogger
-    {
-        IMqttNetChildLogger CreateChildLogger(string source = null);
-
-        void Verbose(string message, params object[] parameters);
-
-        void Info(string message, params object[] parameters);
-
-        void Warning(Exception exception, string message, params object[] parameters);
-
-        void Error(Exception exception, string message, params object[] parameters);
-    }
-}

Source/MQTTnet/Diagnostics/IMqttNetLogger.cs

@@ -6,8 +6,8 @@ namespace MQTTnet.Diagnostics
     {
         event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
 
-        IMqttNetChildLogger CreateChildLogger(string source = null);
+        IMqttNetLogger CreateChildLogger(string source);
 
-        void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception);
+        void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception);
     }
 }

Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs

@@ -1,51 +0,0 @@
-using System;
-
-namespace MQTTnet.Diagnostics
-{
-    public class MqttNetChildLogger : IMqttNetChildLogger
-    {
-        private readonly IMqttNetLogger _logger;
-        private readonly string _source;
-
-        public MqttNetChildLogger(IMqttNetLogger logger, string source)
-        {
-            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
-            _source = source;
-        }
-
-        public IMqttNetChildLogger CreateChildLogger(string source)
-        {
-            string childSource;
-            if (!string.IsNullOrEmpty(_source))
-            {
-                childSource = _source + "." + source;
-            }
-            else
-            {
-                childSource = source;
-            }
-
-            return new MqttNetChildLogger(_logger, childSource);
-        }
-
-        public void Verbose(string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null);
-        }
-
-        public void Info(string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null);
-        }
-
-        public void Warning(Exception exception, string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception);
-        }
-
-        public void Error(Exception exception, string message, params object[] parameters)
-        {
-            _logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception);
-        }
-    }
-}

Source/MQTTnet/Diagnostics/MqttNetLogMessage.cs

@@ -4,30 +4,19 @@ namespace MQTTnet.Diagnostics
 {
     public class MqttNetLogMessage
     {
-        public MqttNetLogMessage(string logId, DateTime timestamp, int threadId, string source, MqttNetLogLevel level, string message, Exception exception)
-        {
-            LogId = logId;
-            Timestamp = timestamp;
-            ThreadId = threadId;
-            Source = source;
-            Level = level;
-            Message = message;
-            Exception = exception;
-        }
+        public string LogId { get; set; }
 
-        public string LogId { get; }
+        public DateTime Timestamp { get; set; }
 
-        public DateTime Timestamp { get; }
+        public int ThreadId { get; set; }
 
-        public int ThreadId { get; }
+        public string Source { get; set; }
 
-        public string Source { get; }
+        public MqttNetLogLevel Level { get; set; }
 
-        public MqttNetLogLevel Level { get; }
+        public string Message { get; set; }
 
-        public string Message { get; }
-
-        public Exception Exception { get; }
+        public Exception Exception { get; set; }
 
         public override string ToString()
         {

Source/MQTTnet/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs

@@ -6,9 +6,14 @@ namespace MQTTnet.Diagnostics
     {
         public MqttNetLogMessagePublishedEventArgs(MqttNetLogMessage logMessage)
         {
+            LogMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage));
+
             TraceMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage));
         }
 
+        [Obsolete("Use new proeprty LogMessage instead.")]
         public MqttNetLogMessage TraceMessage { get; }
+
+        public MqttNetLogMessage LogMessage { get; }
     }
 }

Source/MQTTnet/Diagnostics/MqttNetLogger.cs

@@ -4,26 +4,51 @@ namespace MQTTnet.Diagnostics
 {
     public class MqttNetLogger : IMqttNetLogger
     {
-        private readonly string _logId;
+        readonly string _logId;
+        readonly string _source;
 
-        public MqttNetLogger(string logId = null)
+        readonly MqttNetLogger _parentLogger;
+
+        public MqttNetLogger(string source, string logId)
         {
+            _source = source;
+            _logId = logId;
+        }
+
+        public MqttNetLogger()
+        {
+        }
+
+        public MqttNetLogger(string logId)
+        {
+            _logId = logId;
+        }
+
+        MqttNetLogger(MqttNetLogger parentLogger, string logId, string source)
+        {
+            _parentLogger = parentLogger ?? throw new ArgumentNullException(nameof(parentLogger));
+            _source = source ?? throw new ArgumentNullException(nameof(source));
+
             _logId = logId;
         }
 
         public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
 
-        public IMqttNetChildLogger CreateChildLogger(string source = null)
+        // TODO: Consider creating a LoggerFactory which will allow creating loggers. The logger factory will
+        // be the only place which has the published event.
+        public IMqttNetLogger CreateChildLogger(string source)
         {
-            return new MqttNetChildLogger(this, source);
+            if (source is null) throw new ArgumentNullException(nameof(source));
+
+            return new MqttNetLogger(this, _logId, source);
         }
 
-        public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
+        public void Publish(MqttNetLogLevel level, string message, object[] parameters, Exception exception)
         {
             var hasLocalListeners = LogMessagePublished != null;
             var hasGlobalListeners = MqttNetGlobalLogger.HasListeners;
 
-            if (!hasLocalListeners && !hasGlobalListeners)
+            if (!hasLocalListeners && !hasGlobalListeners && _parentLogger == null)
             {
                 return;
             }
@@ -40,17 +65,35 @@ namespace MQTTnet.Diagnostics
                 }
             }
 
-            var traceMessage = new MqttNetLogMessage(_logId, DateTime.UtcNow, Environment.CurrentManagedThreadId, source, logLevel, message, exception);
+            var logMessage = new MqttNetLogMessage
+            {
+                LogId = _logId,
+                Timestamp = DateTime.UtcNow,
+                Source = _source,
+                ThreadId = Environment.CurrentManagedThreadId,
+                Level = level,
+                Message = message,
+                Exception = exception
+            };
 
             if (hasGlobalListeners)
             {
-                MqttNetGlobalLogger.Publish(traceMessage);
+                MqttNetGlobalLogger.Publish(logMessage);
             }
 
             if (hasLocalListeners)
             {
-                LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(traceMessage));
+                LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage));
             }
+
+            _parentLogger?.Publish(logMessage);
+        }
+
+        void Publish(MqttNetLogMessage logMessage)
+        {
+            LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage));
+
+            _parentLogger?.Publish(logMessage);
         }
     }
 }

Source/MQTTnet/Diagnostics/MqttNetLoggerExtensions.cs

@@ -0,0 +1,35 @@
+using System;
+
+namespace MQTTnet.Diagnostics
+{
+    public static class MqttNetLoggerExtensions
+    {
+        public static void Verbose(this IMqttNetLogger logger, string message, params object[] parameters)
+        {
+            if (logger is null) throw new ArgumentNullException(nameof(logger));
+
+            logger.Publish(MqttNetLogLevel.Verbose, message, parameters, null);
+        }
+
+        public static void Info(this IMqttNetLogger logger, string message, params object[] parameters)
+        {
+            if (logger is null) throw new ArgumentNullException(nameof(logger));
+
+            logger.Publish(MqttNetLogLevel.Info, message, parameters, null);
+        }
+
+        public static void Warning(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters)
+        {
+            if (logger is null) throw new ArgumentNullException(nameof(logger));
+
+            logger.Publish(MqttNetLogLevel.Warning, message, parameters, exception);
+        }
+
+        public static void Error(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters)
+        {
+            if (logger is null) throw new ArgumentNullException(nameof(logger));
+
+            logger.Publish(MqttNetLogLevel.Error, message, parameters, exception);
+        }
+    }
+}

Source/MQTTnet/Diagnostics/TargetFrameworkProvider.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Diagnostics
 {
-    public static class TargetFrameworkInfoProvider
+    public static class TargetFrameworkProvider
     {
         public static string TargetFramework
         {

Source/MQTTnet/Extensions/MqttClientOptionsBuilderExtension.cs

@@ -0,0 +1,48 @@
+using System;
+using System.Linq;
+using MQTTnet.Client.Options;
+
+namespace MQTTnet.Extensions
+{
+    public static class MqttClientOptionsBuilderExtension
+    {
+        public static MqttClientOptionsBuilder WithConnectionUri(this MqttClientOptionsBuilder builder, Uri uri)
+        {
+            var port = uri.IsDefaultPort ? null : (int?) uri.Port;
+            switch (uri.Scheme.ToLower())
+            {
+                case "tcp":
+                case "mqtt":
+                    builder.WithTcpServer(uri.Host, port);
+                    break;
+
+                case "mqtts":
+                    builder.WithTcpServer(uri.Host, port).WithTls();
+                    break;
+
+                case "ws":
+                case "wss":
+                    builder.WithWebSocketServer(uri.ToString());
+                    break;
+
+                default:
+                    throw new ArgumentException("Unexpected scheme in uri.");
+            }
+            
+            if (!string.IsNullOrEmpty(uri.UserInfo))
+            {
+                var userInfo = uri.UserInfo.Split(':');
+                var username = userInfo[0];
+                var password = userInfo.Length > 1 ? userInfo[1] : "";
+                builder.WithCredentials(username, password);
+            }
+
+            return builder;
+        }
+
+        public static MqttClientOptionsBuilder WithConnectionUri(this MqttClientOptionsBuilder builder, string uri)
+        {
+            return WithConnectionUri(builder, new Uri(uri, UriKind.Absolute));
+        }
+    }
+}

Source/MQTTnet/Extensions/UserPropertyExtension.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Linq;
+
+namespace MQTTnet.Extensions
+{
+    public static class UserPropertyExtension
+    {
+        public static string GetUserProperty(this MqttApplicationMessage message, string propertyName, StringComparison comparisonType = StringComparison.Ordinal)
+        {
+            if (message == null) throw new ArgumentNullException(nameof(message));
+            if (propertyName == null) throw new ArgumentNullException(nameof(propertyName));
+
+            return message.UserProperties?.SingleOrDefault(up => up.Name.Equals(propertyName, comparisonType))?.Value;
+        }
+    }
+}

Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs

@@ -20,11 +20,6 @@ namespace MQTTnet.Formatter.V3
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
-            if (applicationMessage.UserProperties?.Any() == true)
-            {
-                throw new MqttProtocolViolationException("User properties are not supported in MQTT version 3.");
-            }
-
             return new MqttPublishPacket
             {
                 Topic = applicationMessage.Topic,
@@ -171,11 +166,6 @@ namespace MQTTnet.Formatter.V3
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            if (options.UserProperties?.Any() == true)
-            {
-                throw new MqttProtocolViolationException("User properties are not supported in MQTT version 3.");
-            }
-
             var subscribePacket = new MqttSubscribePacket();
             subscribePacket.TopicFilters.AddRange(options.TopicFilters);
             
@@ -186,11 +176,6 @@ namespace MQTTnet.Formatter.V3
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            if (options.UserProperties?.Any() == true)
-            {
-                throw new MqttProtocolViolationException("User properties are not supported in MQTT version 3.");
-            }
-
             var unsubscribePacket = new MqttUnsubscribePacket();
             unsubscribePacket.TopicFilters.AddRange(options.TopicFilters);
 

Source/MQTTnet/Implementations/CrossPlatformSocket.cs

@@ -0,0 +1,229 @@
+using System;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Implementations
+{
+    public sealed class CrossPlatformSocket : IDisposable
+    {
+        readonly Socket _socket;
+
+        public CrossPlatformSocket(AddressFamily addressFamily)
+        {
+            _socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
+        }
+
+        public CrossPlatformSocket()
+        {
+            // Having this contructor is important because avoiding the address family as parameter
+            // will make use of dual mode in the .net framework.
+            _socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+        }
+
+        public CrossPlatformSocket(Socket socket)
+        {
+            _socket = socket ?? throw new ArgumentNullException(nameof(socket));
+        }
+
+        public bool NoDelay
+        {
+            get
+            {
+                return (int)_socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay) > 0;
+            }
+
+            set
+            {
+                _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, value ? 1 : 0);
+            }
+        }
+
+        public bool DualMode
+        {
+            get
+            {
+                return _socket.DualMode;
+            }
+
+            set
+            {
+                _socket.DualMode = value;
+            }
+        }
+
+        public int ReceiveBufferSize
+        {
+            get
+            {
+                return _socket.ReceiveBufferSize;
+            }
+
+            set
+            {
+                _socket.ReceiveBufferSize = value;
+            }
+        }
+
+        public int SendBufferSize
+        {
+            get
+            {
+                return _socket.SendBufferSize;
+            }
+
+            set
+            {
+                _socket.SendBufferSize = value;
+            }
+        }
+
+        public EndPoint RemoteEndPoint
+        {
+            get
+            {
+                return _socket.RemoteEndPoint;
+            }
+        }
+
+        public bool ReuseAddress
+        {
+            get
+            {
+                return (int)_socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress) != 0;
+            }
+
+            set
+            {
+                _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, value ? 1 : 0);
+            }
+        }
+
+        public async Task<CrossPlatformSocket> AcceptAsync()
+        {
+            try
+            {
+#if NET452 || NET461
+                var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
+                return new CrossPlatformSocket(clientSocket);
+#else
+                var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
+                return new CrossPlatformSocket(clientSocket);
+#endif
+            }
+            catch (ObjectDisposedException)
+            {
+                // This will happen when _socket.EndAccept gets called by Task library but the socket is already disposed.
+                return null;
+            }
+        }
+
+        public void Bind(EndPoint localEndPoint)
+        {
+            if (localEndPoint is null) throw new ArgumentNullException(nameof(localEndPoint));
+
+            _socket.Bind(localEndPoint);
+        }
+
+        public void Listen(int connectionBacklog)
+        {
+            _socket.Listen(connectionBacklog);
+        }
+
+        public async Task ConnectAsync(string host, int port, CancellationToken cancellationToken)
+        {
+            if (host is null) throw new ArgumentNullException(nameof(host));
+
+            try
+            {
+                // Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430
+                using (cancellationToken.Register(() => _socket.Dispose()))
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+#if NET452 || NET461
+                    await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, host, port, null).ConfigureAwait(false);
+#else
+                    await _socket.ConnectAsync(host, port).ConfigureAwait(false);
+#endif
+                }
+            }
+            catch (ObjectDisposedException)
+            {
+                // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed.
+            }
+        }
+
+        public async Task SendAsync(ArraySegment<byte> buffer, SocketFlags socketFlags)
+        {
+            try
+            {
+#if NET452 || NET461
+                await Task.Factory.FromAsync(SocketWrapper.BeginSend, _socket.EndSend, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false);
+#else
+                await _socket.SendAsync(buffer, socketFlags).ConfigureAwait(false);
+#endif
+            }
+            catch (ObjectDisposedException)
+            {
+                // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed.
+            }
+        }
+
+        public async Task<int> ReceiveAsync(ArraySegment<byte> buffer, SocketFlags socketFlags)
+        {
+            try
+            {
+#if NET452 || NET461
+                return await Task.Factory.FromAsync(SocketWrapper.BeginReceive, _socket.EndReceive, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false);
+#else
+                return await _socket.ReceiveAsync(buffer, socketFlags).ConfigureAwait(false);
+#endif
+            }
+            catch (ObjectDisposedException)
+            {
+                // This will happen when _socket.EndReceive gets called by Task library but the socket is already disposed.
+                return -1;
+            }
+        }
+
+        public NetworkStream GetStream()
+        {
+            return new NetworkStream(_socket, true);
+        }
+
+        public void Dispose()
+        {
+            _socket?.Dispose();
+        }
+
+#if NET452 || NET461
+        class SocketWrapper
+        {
+            readonly Socket _socket;
+            readonly ArraySegment<byte> _buffer;
+            readonly SocketFlags _socketFlags;
+
+            public SocketWrapper(Socket socket, ArraySegment<byte> buffer, SocketFlags socketFlags)
+            {
+                _socket = socket;
+                _buffer = buffer;
+                _socketFlags = socketFlags;
+            }
+
+            public static IAsyncResult BeginSend(AsyncCallback callback, object state)
+            {
+                var socketWrapper = (SocketWrapper)state;
+                return socketWrapper._socket.BeginSend(socketWrapper._buffer.Array, socketWrapper._buffer.Offset, socketWrapper._buffer.Count, socketWrapper._socketFlags, callback, state);
+            }
+
+            public static IAsyncResult BeginReceive(AsyncCallback callback, object state)
+            {
+                var socketWrapper = (SocketWrapper)state;
+                return socketWrapper._socket.BeginReceive(socketWrapper._buffer.Array, socketWrapper._buffer.Offset, socketWrapper._buffer.Count, socketWrapper._socketFlags, callback, state);
+            }
+        }
+#endif
+    }
+}

Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs

@@ -1,17 +1,17 @@
-using System;
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.Client.Options;
 using MQTTnet.Diagnostics;
 using MQTTnet.Formatter;
+using System;
 
 namespace MQTTnet.Implementations
 {
     public class MqttClientAdapterFactory : IMqttClientAdapterFactory
     {
-        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
-            
+
             switch (options.ChannelOptions)
             {
                 case MqttClientTcpOptions _:

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -46,15 +46,15 @@ namespace MQTTnet.Implementations
 
         public async Task ConnectAsync(CancellationToken cancellationToken)
         {
-            Socket socket;
+            CrossPlatformSocket socket;
 
             if (_options.AddressFamily == AddressFamily.Unspecified)
             {
-                socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+                socket = new CrossPlatformSocket();
             }
             else
             {
-                socket = new Socket(_options.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
+                socket = new CrossPlatformSocket(_options.AddressFamily);
             }
 
             socket.ReceiveBufferSize = _options.BufferSize;
@@ -69,20 +69,24 @@ namespace MQTTnet.Implementations
                 socket.DualMode = _options.DualMode.Value;
             }
 
-            // Workaround for: workaround for https://github.com/dotnet/corefx/issues/24430
-            using (cancellationToken.Register(() => socket.Dispose()))
-            {
-                await PlatformAbstractionLayer.ConnectAsync(socket, _options.Server, _options.GetPort()).ConfigureAwait(false);
-            }
+            await socket.ConnectAsync(_options.Server, _options.GetPort(), cancellationToken).ConfigureAwait(false);
 
-            var networkStream = new NetworkStream(socket, true);
+            var networkStream = socket.GetStream();
 
             if (_options.TlsOptions.UseTls)
             {
                 var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
-                _stream = sslStream;
+                try
+                {
+                    await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+                }
+                catch
+                {
+                    sslStream.Dispose();
+                    throw;
+                }
 
-                await sslStream.AuthenticateAsClientAsync(_options.Server, LoadCertificates(), _options.TlsOptions.SslProtocol, !_options.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
+                _stream = sslStream;
             }
             else
             {
@@ -107,17 +111,14 @@ namespace MQTTnet.Implementations
                 // Workaround for: https://github.com/dotnet/corefx/issues/24430
                 using (cancellationToken.Register(Dispose))
                 {
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return 0;
-                    }
+                    cancellationToken.ThrowIfCancellationRequested();
 
                     return await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (ObjectDisposedException)
             {
-                return 0;
+                return -1;
             }
             catch (IOException exception)
             {
@@ -139,10 +140,7 @@ namespace MQTTnet.Implementations
                 // Workaround for: https://github.com/dotnet/corefx/issues/24430
                 using (cancellationToken.Register(Dispose))
                 {
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return;
-                    }
+                    cancellationToken.ThrowIfCancellationRequested();
 
                     await _stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
                 }

Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs

@@ -11,14 +11,14 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttTcpServerAdapter : IMqttServerAdapter
+    public sealed class MqttTcpServerAdapter : IMqttServerAdapter
     {
-        private readonly IMqttNetChildLogger _logger;
+        readonly IMqttNetLogger _logger;
 
-        private IMqttServerOptions _options;
-        private StreamSocketListener _listener;
+        IMqttServerOptions _options;
+        StreamSocketListener _listener;
 
-        public MqttTcpServerAdapter(IMqttNetChildLogger logger)
+        public MqttTcpServerAdapter(IMqttNetLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
@@ -68,7 +68,7 @@ namespace MQTTnet.Implementations
             _listener = null;
         }
 
-        private async void OnConnectionReceivedAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
+        async void OnConnectionReceivedAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
         {
             try
             {

Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs

@@ -1,4 +1,8 @@
 #if !WINDOWS_UWP
+using MQTTnet.Adapter;
+using MQTTnet.Diagnostics;
+using MQTTnet.Internal;
+using MQTTnet.Server;
 using System;
 using System.Collections.Generic;
 using System.Net;
@@ -6,21 +10,17 @@ using System.Net.Sockets;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
-using MQTTnet.Internal;
-using MQTTnet.Server;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttTcpServerAdapter : Disposable, IMqttServerAdapter
+    public sealed class MqttTcpServerAdapter : Disposable, IMqttServerAdapter
     {
-        private readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
-        private readonly IMqttNetChildLogger _logger;
+        readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
+        readonly IMqttNetLogger _logger;
 
-        private CancellationTokenSource _cancellationTokenSource;
+        CancellationTokenSource _cancellationTokenSource;
 
-        public MqttTcpServerAdapter(IMqttNetChildLogger logger)
+        public MqttTcpServerAdapter(IMqttNetLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
@@ -59,7 +59,7 @@ namespace MQTTnet.Implementations
                 {
                     tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate, options.TlsEndpointOptions.CertificateCredentials.Password);
                 }
-                
+
                 if (!tlsCertificate.HasPrivateKey)
                 {
                     throw new InvalidOperationException("The certificate for TLS encryption must contain the private key.");
@@ -77,7 +77,17 @@ namespace MQTTnet.Implementations
             return Task.FromResult(0);
         }
 
-        private void Cleanup()
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                Cleanup();
+            }
+
+            base.Dispose(disposing);
+        }
+
+        void Cleanup()
         {
             _cancellationTokenSource?.Cancel(false);
             _cancellationTokenSource?.Dispose();
@@ -91,16 +101,7 @@ namespace MQTTnet.Implementations
             _listeners.Clear();
         }
 
-        protected override void Dispose(bool disposing)
-        {
-            if (disposing)
-            {
-                Cleanup();
-            }
-            base.Dispose(disposing);
-        }
-
-        private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken)
+        void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken)
         {
             if (!options.BoundInterNetworkAddress.Equals(IPAddress.None))
             {

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -1,4 +1,9 @@
 #if !WINDOWS_UWP
+using MQTTnet.Adapter;
+using MQTTnet.Diagnostics;
+using MQTTnet.Formatter;
+using MQTTnet.Internal;
+using MQTTnet.Server;
 using System;
 using System.IO;
 using System.Net;
@@ -7,30 +12,25 @@ using System.Net.Sockets;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
-using MQTTnet.Formatter;
-using MQTTnet.Internal;
-using MQTTnet.Server;
 
 namespace MQTTnet.Implementations
 {
-    public class MqttTcpServerListener : IDisposable
+    public sealed class MqttTcpServerListener : IDisposable
     {
-        private readonly IMqttNetChildLogger _logger;
-        private readonly AddressFamily _addressFamily;
-        private readonly MqttServerTcpEndpointBaseOptions _options;
-        private readonly MqttServerTlsTcpEndpointOptions _tlsOptions;
-        private readonly X509Certificate2 _tlsCertificate;
+        readonly IMqttNetLogger _logger;
+        readonly AddressFamily _addressFamily;
+        readonly MqttServerTcpEndpointBaseOptions _options;
+        readonly MqttServerTlsTcpEndpointOptions _tlsOptions;
+        readonly X509Certificate2 _tlsCertificate;
 
-        private Socket _socket;
+        private CrossPlatformSocket _socket;
         private IPEndPoint _localEndPoint;
 
         public MqttTcpServerListener(
             AddressFamily addressFamily,
             MqttServerTcpEndpointBaseOptions options,
             X509Certificate2 tlsCertificate,
-            IMqttNetChildLogger logger)
+            IMqttNetLogger logger)
         {
             _addressFamily = addressFamily;
             _options = options;
@@ -59,20 +59,20 @@ namespace MQTTnet.Implementations
 
                 _logger.Info($"Starting TCP listener for {_localEndPoint} TLS={_tlsCertificate != null}.");
 
-                _socket = new Socket(_addressFamily, SocketType.Stream, ProtocolType.Tcp);
+                _socket = new CrossPlatformSocket(_addressFamily);
 
                 // Usage of socket options is described here: https://docs.microsoft.com/en-us/dotnet/api/system.net.sockets.socket.setsocketoption?view=netcore-2.2
 
                 if (_options.ReuseAddress)
                 {
-                    _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
+                    _socket.ReuseAddress = true;
                 }
-                
+
                 if (_options.NoDelay)
                 {
-                    _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, true);
+                    _socket.NoDelay = true;
                 }
-                
+
                 _socket.Bind(_localEndPoint);
                 _socket.Listen(_options.ConnectionBacklog);
 
@@ -87,7 +87,7 @@ namespace MQTTnet.Implementations
                     throw;
                 }
 
-                _logger.Warning(exception,"Error while creating listener socket for local end point '{0}'.", _localEndPoint);
+                _logger.Warning(exception, "Error while creating listener socket for local end point '{0}'.", _localEndPoint);
                 return false;
             }
         }
@@ -101,13 +101,13 @@ namespace MQTTnet.Implementations
 #endif
         }
 
-        private async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken)
+        async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken)
         {
             while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
-                    var clientSocket = await PlatformAbstractionLayer.AcceptAsync(_socket).ConfigureAwait(false);
+                    var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
                     if (clientSocket == null)
                     {
                         continue;
@@ -116,7 +116,7 @@ namespace MQTTnet.Implementations
                     Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken).Forget(_logger);
                 }
                 catch (OperationCanceledException)
-                {  
+                {
                 }
                 catch (Exception exception)
                 {
@@ -128,14 +128,14 @@ namespace MQTTnet.Implementations
                             continue;
                         }
                     }
-                   
+
                     _logger.Error(exception, $"Error while accepting connection at TCP listener {_localEndPoint} TLS={_tlsCertificate != null}.");
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
                 }
             }
         }
 
-        private async Task TryHandleClientConnectionAsync(Socket clientSocket)
+        async Task TryHandleClientConnectionAsync(CrossPlatformSocket clientSocket)
         {
             Stream stream = null;
             string remoteEndPoint = null;
@@ -151,7 +151,7 @@ namespace MQTTnet.Implementations
 
                 clientSocket.NoDelay = _options.NoDelay;
 
-                stream = new NetworkStream(clientSocket, true);
+                stream = clientSocket.GetStream();
 
                 X509Certificate2 clientCertificate = null;
 
@@ -160,9 +160,9 @@ namespace MQTTnet.Implementations
                     var sslStream = new SslStream(stream, false, _tlsOptions.RemoteCertificateValidationCallback);
 
                     await sslStream.AuthenticateAsServerAsync(
-                        _tlsCertificate, 
-                        _tlsOptions.ClientCertificateRequired, 
-                        _tlsOptions.SslProtocol, 
+                        _tlsCertificate,
+                        _tlsOptions.ClientCertificateRequired,
+                        _tlsOptions.SslProtocol,
                         _tlsOptions.CheckCertificateRevocation).ConfigureAwait(false);
 
                     stream = sslStream;

Source/MQTTnet/Implementations/PlatformAbstractionLayer.cs

@@ -1,94 +1,9 @@
-using System;
-using System.Net;
-using System.Net.Sockets;
-using System.Threading.Tasks;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Implementations
 {
     public static class PlatformAbstractionLayer
     {
-        // TODO: Consider creating primitives like "MqttNetSocket" which will wrap all required methods and do the platform stuff.
-        public static async Task<Socket> AcceptAsync(Socket socket)
-        {
-#if NET452 || NET461
-            try
-            {
-                return await Task.Factory.FromAsync(socket.BeginAccept, socket.EndAccept, null).ConfigureAwait(false);
-            }
-            catch (ObjectDisposedException)
-            {
-                return null;
-            }
-#else
-            return await socket.AcceptAsync().ConfigureAwait(false);
-#endif
-        }
-
-
-        public static Task ConnectAsync(Socket socket, IPAddress ip, int port)
-        {
-#if NET452 || NET461
-            return Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, ip, port, null);
-#else
-            return socket.ConnectAsync(ip, port);
-#endif
-        }
-
-        public static Task ConnectAsync(Socket socket, string host, int port)
-        {
-#if NET452 || NET461
-            return Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, host, port, null);
-#else
-            return socket.ConnectAsync(host, port);
-#endif
-        }
-
-#if NET452 || NET461
-        public class SocketWrapper 
-        {
-            private readonly Socket _socket;
-            private readonly ArraySegment<byte> _buffer;
-            private readonly SocketFlags _socketFlags;
-
-            public SocketWrapper(Socket socket, ArraySegment<byte> buffer, SocketFlags socketFlags)
-            {
-                _socket = socket;
-                _buffer = buffer;
-                _socketFlags = socketFlags;
-            }
-
-            public static IAsyncResult BeginSend(AsyncCallback callback, object state)
-            {
-                var real = (SocketWrapper)state;
-                return real._socket.BeginSend(real._buffer.Array, real._buffer.Offset, real._buffer.Count, real._socketFlags, callback, state);
-            }
-
-            public static IAsyncResult BeginReceive(AsyncCallback callback, object state)
-            {
-                var real = (SocketWrapper)state;
-                return real._socket.BeginReceive(real._buffer.Array, real._buffer.Offset, real._buffer.Count, real._socketFlags, callback, state);
-            }
-        }
-#endif
-
-        public static Task SendAsync(Socket socket, ArraySegment<byte> buffer, SocketFlags socketFlags)
-        {
-#if NET452 || NET461            
-            return Task.Factory.FromAsync(SocketWrapper.BeginSend, socket.EndSend, new SocketWrapper(socket, buffer, socketFlags));
-#else
-            return socket.SendAsync(buffer, socketFlags);
-#endif
-        }
-
-        public static Task<int> ReceiveAsync(Socket socket, ArraySegment<byte> buffer, SocketFlags socketFlags)
-        {
-#if NET452 || NET461
-            return Task.Factory.FromAsync(SocketWrapper.BeginReceive, socket.EndReceive, new SocketWrapper(socket, buffer, socketFlags));
-#else
-            return socket.ReceiveAsync(buffer, socketFlags);
-#endif
-        }
-
         public static Task CompletedTask
         {
             get

Source/MQTTnet/Internal/TaskExtensions.cs

@@ -1,11 +1,11 @@
-using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
+using MQTTnet.Diagnostics;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Internal
 {
     public static class TaskExtensions
     {
-        public static void Forget(this Task task, IMqttNetChildLogger logger)
+        public static void Forget(this Task task, IMqttNetLogger logger)
         {
             task?.ContinueWith(t =>
                 {

Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs

@@ -0,0 +1,19 @@
+using MQTTnet.Client.Options;
+using MQTTnet.Packets;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.LowLevelClient
+{
+    public interface ILowLevelMqttClient : IDisposable
+    {
+        Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken);
+
+        Task DisconnectAsync(CancellationToken cancellationToken);
+
+        Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken);
+
+        Task<MqttBasePacket> ReceiveAsync(CancellationToken cancellationToken);
+    }
+}

Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs

@@ -0,0 +1,128 @@
+using MQTTnet.Adapter;
+using MQTTnet.Client.Options;
+using MQTTnet.Diagnostics;
+using MQTTnet.Packets;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.LowLevelClient
+{
+    public sealed class LowLevelMqttClient : ILowLevelMqttClient
+    {
+        readonly IMqttNetLogger _logger;
+        readonly IMqttClientAdapterFactory _clientAdapterFactory;
+
+        IMqttChannelAdapter _adapter;
+        IMqttClientOptions _options;
+
+        public LowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory, IMqttNetLogger logger)
+        {
+            if (clientAdapterFactory is null) throw new ArgumentNullException(nameof(clientAdapterFactory));
+            if (logger is null) throw new ArgumentNullException(nameof(logger));
+
+            _clientAdapterFactory = clientAdapterFactory;
+            _logger = logger.CreateChildLogger(nameof(LowLevelMqttClient));
+        }
+
+        bool IsConnected => _adapter != null;
+
+        public async Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken)
+        {
+            if (options is null) throw new ArgumentNullException(nameof(options));
+
+            if (_adapter != null)
+            {
+                throw new InvalidOperationException("Low level MQTT client is already connected. Disconnect first before connecting again.");
+            }
+
+            var newAdapter = _clientAdapterFactory.CreateClientAdapter(options, _logger);
+
+            try
+            {
+                _logger.Verbose($"Trying to connect with server '{options.ChannelOptions}' (Timeout={options.CommunicationTimeout}).");
+                await newAdapter.ConnectAsync(options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+                _logger.Verbose("Connection with server established.");
+
+                _options = options;
+            }
+            catch (Exception)
+            {
+                _adapter.Dispose();
+                throw;
+            }
+
+            _adapter = newAdapter;
+        }
+
+        public async Task DisconnectAsync(CancellationToken cancellationToken)
+        {
+            if (_adapter == null)
+            {
+                return;
+            }
+
+            await SafeDisconnect(cancellationToken).ConfigureAwait(false);
+            _adapter = null;
+        }
+
+        public async Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken)
+        {
+            if (packet is null) throw new ArgumentNullException(nameof(packet));
+
+            if (_adapter == null)
+            {
+                throw new InvalidOperationException("Low level MQTT client is not connected.");
+            }
+
+            try
+            {
+                await _adapter.SendPacketAsync(packet, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception)
+            {
+                await SafeDisconnect(cancellationToken).ConfigureAwait(false);
+                throw;
+            }
+        }
+
+        public async Task<MqttBasePacket> ReceiveAsync(CancellationToken cancellationToken)
+        {
+            if (_adapter == null)
+            {
+                throw new InvalidOperationException("Low level MQTT client is not connected.");
+            }
+
+            try
+            {
+                return await _adapter.ReceivePacketAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception)
+            {
+                await SafeDisconnect(cancellationToken).ConfigureAwait(false);
+                throw;
+            }
+        }
+
+        public void Dispose()
+        {
+            _adapter?.Dispose();
+        }
+
+        async Task SafeDisconnect(CancellationToken cancellationToken)
+        {
+            try
+            {
+                await _adapter.DisconnectAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error while disconnecting.");
+            }
+            finally
+            {
+                _adapter.Dispose();
+            }
+        }
+    }
+}

Source/MQTTnet/MQTTnet.csproj

@@ -64,7 +64,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.9" />
+    <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
   </ItemGroup>
   
 </Project>

Source/MQTTnet/MqttFactory.cs

@@ -1,16 +1,17 @@
-using System;
-using System.Collections.Generic;
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Implementations;
+using MQTTnet.LowLevelClient;
 using MQTTnet.Server;
+using System;
+using System.Collections.Generic;
 
 namespace MQTTnet
 {
-    public class MqttFactory : IMqttFactory
+    public sealed class MqttFactory : IMqttFactory
     {
-        private IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory();
+        IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory();
 
         public MqttFactory() : this(new MqttNetLogger())
         {
@@ -29,6 +30,33 @@ namespace MQTTnet
             return this;
         }
 
+        public ILowLevelMqttClient CreateLowLevelMqttClient()
+        {
+            return CreateLowLevelMqttClient(DefaultLogger);
+        }
+
+        public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger)
+        {
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
+
+            return new LowLevelMqttClient(_clientAdapterFactory, logger);
+        }
+
+        public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory)
+        {
+            if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory));
+
+            return new LowLevelMqttClient(_clientAdapterFactory, DefaultLogger);
+        }
+
+        public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactoryy)
+        {
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            if (clientAdapterFactoryy == null) throw new ArgumentNullException(nameof(clientAdapterFactoryy));
+
+            return new LowLevelMqttClient(_clientAdapterFactory, logger);
+        }
+
         public IMqttClient CreateMqttClient()
         {
             return CreateMqttClient(DefaultLogger);
@@ -65,7 +93,7 @@ namespace MQTTnet
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
-            return CreateMqttServer(new List<IMqttServerAdapter> { new MqttTcpServerAdapter(logger.CreateChildLogger()) }, logger);
+            return CreateMqttServer(new List<IMqttServerAdapter> { new MqttTcpServerAdapter(logger) }, logger);
         }
 
         public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters, IMqttNetLogger logger)
@@ -73,14 +101,14 @@ namespace MQTTnet
             if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters));
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
-            return new MqttServer(serverAdapters, logger.CreateChildLogger());
+            return new MqttServer(serverAdapters, logger);
         }
 
         public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters)
         {
             if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters));
-            
-            return new MqttServer(serverAdapters, DefaultLogger.CreateChildLogger());
+
+            return new MqttServer(serverAdapters, DefaultLogger);
         }
     }
 }

Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs

@@ -1,12 +1,12 @@
-using System.Collections.Generic;
+using MQTTnet.Diagnostics;
+using System.Collections.Generic;
 using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
 
 namespace MQTTnet.Server
 {
     public interface IMqttRetainedMessagesManager
     {
-        Task Start(IMqttServerOptions options, IMqttNetChildLogger logger);
+        Task Start(IMqttServerOptions options, IMqttNetLogger logger);
 
         Task LoadMessagesAsync();
 

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -3,6 +3,7 @@ using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
+using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.PacketDispatcher;
 using MQTTnet.Packets;
@@ -15,32 +16,34 @@ using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
-    public class MqttClientConnection : IDisposable
+    public sealed class MqttClientConnection : IDisposable
     {
-        private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
-        private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
-        private readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
+        readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
+        readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
+        readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();
 
-        private readonly IMqttRetainedMessagesManager _retainedMessagesManager;
-        private readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
-        private readonly MqttClientSessionsManager _sessionsManager;
+        readonly IMqttRetainedMessagesManager _retainedMessagesManager;
+        readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
+        readonly MqttClientSessionsManager _sessionsManager;
 
-        private readonly IMqttNetChildLogger _logger;
-        private readonly IMqttServerOptions _serverOptions;
+        readonly IMqttNetLogger _logger;
+        readonly IMqttServerOptions _serverOptions;
 
-        private readonly IMqttChannelAdapter _channelAdapter;
-        private readonly IMqttDataConverter _dataConverter;
-        private readonly string _endpoint;
-        private readonly DateTime _connectedTimestamp;
+        readonly IMqttChannelAdapter _channelAdapter;
+        readonly IMqttDataConverter _dataConverter;
+        readonly string _endpoint;
+        readonly DateTime _connectedTimestamp;
 
-        private Task<MqttClientDisconnectType> _packageReceiverTask;
-        private DateTime _lastPacketReceivedTimestamp;
-        private DateTime _lastNonKeepAlivePacketReceivedTimestamp;
+        Task<MqttClientDisconnectType> _packageReceiverTask;
+        DateTime _lastPacketReceivedTimestamp;
+        DateTime _lastNonKeepAlivePacketReceivedTimestamp;
 
-        private long _receivedPacketsCount;
-        private long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere.
-        private long _receivedApplicationMessagesCount;
-        private long _sentApplicationMessagesCount;
+        long _receivedPacketsCount;
+        long _sentPacketsCount = 1; // Start with 1 because the CONNECT packet is not counted anywhere.
+        long _receivedApplicationMessagesCount;
+        long _sentApplicationMessagesCount;
+
+        bool _isTakeover;
 
         public MqttClientConnection(
             MqttConnectPacket connectPacket,
@@ -49,7 +52,7 @@ namespace MQTTnet.Server
             IMqttServerOptions serverOptions,
             MqttClientSessionsManager sessionsManager,
             IMqttRetainedMessagesManager retainedMessagesManager,
-            IMqttNetChildLogger logger)
+            IMqttNetLogger logger)
         {
             Session = session ?? throw new ArgumentNullException(nameof(session));
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
@@ -64,7 +67,7 @@ namespace MQTTnet.Server
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttClientConnection));
 
-            _keepAliveMonitor = new MqttClientKeepAliveMonitor(ConnectPacket.ClientId, StopAsync, _logger);
+            _keepAliveMonitor = new MqttClientKeepAliveMonitor(ConnectPacket.ClientId, () => StopAsync(), _logger);
 
             _connectedTimestamp = DateTime.UtcNow;
             _lastPacketReceivedTimestamp = _connectedTimestamp;
@@ -77,15 +80,21 @@ namespace MQTTnet.Server
 
         public MqttClientSession Session { get; }
 
-        public async Task StopAsync()
+        public bool IsFinalized { get; set; }
+
+        public Task StopAsync(bool isTakeover = false)
         {
+            _isTakeover = isTakeover;
+
             StopInternal();
 
             var task = _packageReceiverTask;
             if (task != null)
             {
-                await task.ConfigureAwait(false);
+                return task;
             }
+
+            return PlatformAbstractionLayer.CompletedTask;
         }
 
         public void ResetStatistics()
@@ -124,7 +133,7 @@ namespace MQTTnet.Server
             return _packageReceiverTask;
         }
 
-        private async Task<MqttClientDisconnectType> RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext)
+        async Task<MqttClientDisconnectType> RunInternalAsync(MqttConnectionValidatorContext connectionValidatorContext)
         {
             var disconnectType = MqttClientDisconnectType.NotClean;
             try
@@ -243,20 +252,25 @@ namespace MQTTnet.Server
                 _channelAdapter.ReadingPacketStartedCallback = null;
                 _channelAdapter.ReadingPacketCompletedCallback = null;
 
-                _logger.Info("Client '{0}': Session stopped.", ClientId);
+                _logger.Info("Client '{0}': Connection stopped.", ClientId);
 
                 _packageReceiverTask = null;
             }
 
+            if (_isTakeover)
+            {
+                return MqttClientDisconnectType.Takeover;
+            }
+
             return disconnectType;
         }
 
-        private void StopInternal()
+        void StopInternal()
         {
             _cancellationToken.Cancel(false);
         }
 
-        private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
+        async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters)
         {
             var retainedMessages = await _retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
             foreach (var applicationMessage in retainedMessages)
@@ -265,7 +279,7 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket)
+        async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket)
         {
             // TODO: Let the channel adapter create the packet.
             var subscribeResult = await Session.SubscriptionsManager.SubscribeAsync(subscribePacket, ConnectPacket).ConfigureAwait(false);
@@ -281,14 +295,14 @@ namespace MQTTnet.Server
             await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false);
         }
 
-        private async Task HandleIncomingUnsubscribePacketAsync(MqttUnsubscribePacket unsubscribePacket)
+        async Task HandleIncomingUnsubscribePacketAsync(MqttUnsubscribePacket unsubscribePacket)
         {
             // TODO: Let the channel adapter create the packet.
             var unsubscribeResult = await Session.SubscriptionsManager.UnsubscribeAsync(unsubscribePacket).ConfigureAwait(false);
             await SendAsync(unsubscribeResult).ConfigureAwait(false);
         }
 
-        private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
+        Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
         {
             Interlocked.Increment(ref _sentApplicationMessagesCount);
 
@@ -313,16 +327,16 @@ namespace MQTTnet.Server
             }
         }
 
-        private Task HandleIncomingPublishPacketWithQoS0Async(MqttPublishPacket publishPacket)
+        Task HandleIncomingPublishPacketWithQoS0Async(MqttPublishPacket publishPacket)
         {
             var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket);
 
             _sessionsManager.DispatchApplicationMessage(applicationMessage, this);
 
-            return Task.FromResult(0);
+            return PlatformAbstractionLayer.CompletedTask;
         }
 
-        private Task HandleIncomingPublishPacketWithQoS1Async(MqttPublishPacket publishPacket)
+        Task HandleIncomingPublishPacketWithQoS1Async(MqttPublishPacket publishPacket)
         {
             var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket);
             _sessionsManager.DispatchApplicationMessage(applicationMessage, this);
@@ -331,7 +345,7 @@ namespace MQTTnet.Server
             return SendAsync(pubAckPacket);
         }
 
-        private Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket)
+        Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket)
         {
             var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket);
             _sessionsManager.DispatchApplicationMessage(applicationMessage, this);
@@ -345,7 +359,7 @@ namespace MQTTnet.Server
             return SendAsync(pubRecPacket);
         }
 
-        private async Task SendPendingPacketsAsync(CancellationToken cancellationToken)
+        async Task SendPendingPacketsAsync(CancellationToken cancellationToken)
         {
             MqttQueuedApplicationMessage queuedApplicationMessage = null;
             MqttPublishPacket publishPacket = null;
@@ -422,9 +436,6 @@ namespace MQTTnet.Server
                     }
 
                     _logger.Verbose("Queued application message sent (ClientId: {0}).", ClientId);
-
-                    // TODO:
-                    //Interlocked.Increment(ref _sentPacketsCount);
                 }
             }
             catch (Exception exception)
@@ -459,7 +470,7 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task SendAsync(MqttBasePacket packet)
+        async Task SendAsync(MqttBasePacket packet)
         {
             await _channelAdapter.SendPacketAsync(packet, _serverOptions.DefaultCommunicationTimeout, _cancellationToken.Token).ConfigureAwait(false);
 
@@ -471,12 +482,12 @@ namespace MQTTnet.Server
             }
         }
 
-        private void OnAdapterReadingPacketCompleted()
+        void OnAdapterReadingPacketCompleted()
         {
             _keepAliveMonitor?.Resume();
         }
 
-        private void OnAdapterReadingPacketStarted()
+        void OnAdapterReadingPacketStarted()
         {
             _keepAliveMonitor?.Pause();
         }

Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs

@@ -1,27 +1,27 @@
-using System;
+using MQTTnet.Diagnostics;
+using MQTTnet.Internal;
+using System;
 using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
-using MQTTnet.Internal;
 
 namespace MQTTnet.Server
 {
     public class MqttClientKeepAliveMonitor
     {
-        private readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
+        readonly Stopwatch _lastPacketReceivedTracker = new Stopwatch();
 
-        private readonly string _clientId;
-        private readonly Func<Task> _keepAliveElapsedCallback;
-        private readonly IMqttNetChildLogger _logger;
+        readonly string _clientId;
+        readonly Func<Task> _keepAliveElapsedCallback;
+        readonly IMqttNetLogger _logger;
 
-        private bool _isPaused;
+        bool _isPaused;
 
-        public MqttClientKeepAliveMonitor(string clientId, Func<Task> keepAliveElapsedCallback, IMqttNetChildLogger logger)
+        public MqttClientKeepAliveMonitor(string clientId, Func<Task> keepAliveElapsedCallback, IMqttNetLogger logger)
         {
             _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
             _keepAliveElapsedCallback = keepAliveElapsedCallback ?? throw new ArgumentNullException(nameof(keepAliveElapsedCallback));
-            
+
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor));
         }
@@ -32,7 +32,7 @@ namespace MQTTnet.Server
             {
                 return;
             }
-            
+
             Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken).Forget(_logger);
         }
 
@@ -51,7 +51,7 @@ namespace MQTTnet.Server
             _lastPacketReceivedTracker.Restart();
         }
 
-        private async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken)
+        async Task RunAsync(int keepAlivePeriod, CancellationToken cancellationToken)
         {
             try
             {

Source/MQTTnet/Server/MqttClientSession.cs

@@ -1,18 +1,18 @@
-using System;
+using MQTTnet.Diagnostics;
+using MQTTnet.Server.Status;
+using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
-using MQTTnet.Server.Status;
 
 namespace MQTTnet.Server
 {
     public class MqttClientSession
     {
-        private readonly IMqttNetChildLogger _logger;
+        readonly IMqttNetLogger _logger;
 
-        private readonly DateTime _createdTimestamp = DateTime.UtcNow;
+        readonly DateTime _createdTimestamp = DateTime.UtcNow;
 
-        public MqttClientSession(string clientId, IDictionary<object, object> items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetChildLogger logger)
+        public MqttClientSession(string clientId, IDictionary<object, object> items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetLogger logger)
         {
             ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
             Items = items ?? throw new ArgumentNullException(nameof(items));
@@ -73,6 +73,7 @@ namespace MQTTnet.Server
             status.ClientId = ClientId;
             status.CreatedTimestamp = _createdTimestamp;
             status.PendingApplicationMessagesCount = ApplicationMessagesQueue.Count;
+            status.Items = Items;
         }
     }
 }

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -16,26 +16,26 @@ namespace MQTTnet.Server
 {
     public class MqttClientSessionsManager : Disposable
     {
-        private readonly AsyncQueue<MqttEnqueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttEnqueuedApplicationMessage>();
+        readonly AsyncQueue<MqttEnqueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttEnqueuedApplicationMessage>();
 
-        private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1);
-        private readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
-        private readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
-        private readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
+        readonly AsyncLock _createConnectionGate = new AsyncLock();
+        readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
+        readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
+        readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
 
-        private readonly CancellationToken _cancellationToken;
-        private readonly MqttServerEventDispatcher _eventDispatcher;
+        readonly CancellationToken _cancellationToken;
+        readonly MqttServerEventDispatcher _eventDispatcher;
 
-        private readonly IMqttRetainedMessagesManager _retainedMessagesManager;
-        private readonly IMqttServerOptions _options;
-        private readonly IMqttNetChildLogger _logger;
+        readonly IMqttRetainedMessagesManager _retainedMessagesManager;
+        readonly IMqttServerOptions _options;
+        readonly IMqttNetLogger _logger;
 
         public MqttClientSessionsManager(
             IMqttServerOptions options,
             IMqttRetainedMessagesManager retainedMessagesManager,
             CancellationToken cancellationToken,
             MqttServerEventDispatcher eventDispatcher,
-            IMqttNetChildLogger logger)
+            IMqttNetLogger logger)
         {
             _cancellationToken = cancellationToken;
 
@@ -60,9 +60,11 @@ namespace MQTTnet.Server
             }
         }
 
-        public Task HandleClientAsync(IMqttChannelAdapter clientAdapter)
+        public Task HandleClientConnectionAsync(IMqttChannelAdapter clientAdapter)
         {
-            return HandleClientAsync(clientAdapter, _cancellationToken);
+            if (clientAdapter is null) throw new ArgumentNullException(nameof(clientAdapter));
+
+            return HandleClientConnectionAsync(clientAdapter, _cancellationToken);
         }
 
         public Task<IList<IMqttClientStatus>> GetClientStatusAsync()
@@ -155,7 +157,7 @@ namespace MQTTnet.Server
             base.Dispose(disposing);
         }
 
-        private async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)
+        async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)
         {
             while (!cancellationToken.IsCancellationRequested)
             {
@@ -173,7 +175,7 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task TryProcessNextQueuedApplicationMessageAsync(CancellationToken cancellationToken)
+        async Task TryProcessNextQueuedApplicationMessageAsync(CancellationToken cancellationToken)
         {
             try
             {
@@ -207,7 +209,7 @@ namespace MQTTnet.Server
                     applicationMessage = interceptorContext.ApplicationMessage;
                 }
 
-                await _eventDispatcher.HandleApplicationMessageReceivedAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
 
                 if (applicationMessage.Retain)
                 {
@@ -231,14 +233,14 @@ namespace MQTTnet.Server
             }
         }
 
-        private async Task HandleClientAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
+        async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
         {
             var disconnectType = MqttClientDisconnectType.NotClean;
             string clientId = null;
-            var clientWasConnected = true;
-
-            MqttConnectPacket connectPacket = null;
+            var clientWasAuthorized = false;
 
+            MqttConnectPacket connectPacket;
+            MqttClientConnection clientConnection = null;
             try
             {
                 try
@@ -259,11 +261,8 @@ namespace MQTTnet.Server
 
                 var connectionValidatorContext = await ValidateConnectionAsync(connectPacket, channelAdapter).ConfigureAwait(false);
 
-                clientId = connectPacket.ClientId;
-
                 if (connectionValidatorContext.ReasonCode != MqttConnectReasonCode.Success)
                 {
-                    clientWasConnected = false;
                     // Send failure response here without preparing a session. The result for a successful connect
                     // will be sent from the session itself.
                     var connAckPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnAckPacket(connectionValidatorContext);
@@ -272,11 +271,13 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                var connection = await CreateConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
+                clientWasAuthorized = true;
+                clientId = connectPacket.ClientId;
+                clientConnection = await CreateClientConnectionAsync(connectPacket, connectionValidatorContext, channelAdapter).ConfigureAwait(false);
 
-                await _eventDispatcher.HandleClientConnectedAsync(clientId).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientConnectedAsync(clientId).ConfigureAwait(false);
 
-                disconnectType = await connection.RunAsync(connectionValidatorContext).ConfigureAwait(false);
+                disconnectType = await clientConnection.RunAsync(connectionValidatorContext).ConfigureAwait(false);
             }
             catch (OperationCanceledException)
             {
@@ -287,8 +288,10 @@ namespace MQTTnet.Server
             }
             finally
             {
-                if (clientWasConnected)
+                if (clientWasAuthorized && disconnectType != MqttClientDisconnectType.Takeover)
                 {
+                    // Only cleanup if the client was authorized. If not it will remove the existing connection, session etc.
+                    // This allows to kill connections and sessions from known client IDs.
                     if (clientId != null)
                     {
                         _connections.TryRemove(clientId, out _);
@@ -298,18 +301,23 @@ namespace MQTTnet.Server
                             await DeleteSessionAsync(clientId).ConfigureAwait(false);
                         }
                     }
+                }
 
-                    await TryCleanupChannelAsync(channelAdapter).ConfigureAwait(false);
+                await SafeCleanupChannelAsync(channelAdapter).ConfigureAwait(false);
 
-                    if (clientId != null)
-                    {
-                        await _eventDispatcher.TryHandleClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false);
-                    }
+                if (clientWasAuthorized && clientId != null)
+                {
+                    await _eventDispatcher.SafeNotifyClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false);
+                }
+
+                if (clientConnection != null)
+                {
+                    clientConnection.IsFinalized = true;
                 }
             }
         }
 
-        private async Task<MqttConnectionValidatorContext> ValidateConnectionAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
+        async Task<MqttConnectionValidatorContext> ValidateConnectionAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
         {
             var context = new MqttConnectionValidatorContext(connectPacket, channelAdapter, new ConcurrentDictionary<object, object>());
 
@@ -337,17 +345,22 @@ namespace MQTTnet.Server
             return context;
         }
 
-        private async Task<MqttClientConnection> CreateConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
+        async Task<MqttClientConnection> CreateClientConnectionAsync(MqttConnectPacket connectPacket, MqttConnectionValidatorContext connectionValidatorContext, IMqttChannelAdapter channelAdapter)
         {
-            await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false);
-            try
+            using (await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false))
             {
                 var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var session);
 
                 var isConnectionPresent = _connections.TryGetValue(connectPacket.ClientId, out var existingConnection);
                 if (isConnectionPresent)
                 {
-                    await existingConnection.StopAsync().ConfigureAwait(false);
+                    await existingConnection.StopAsync(true);
+
+                    // TODO: This fixes a race condition with unit test Same_Client_Id_Connect_Disconnect_Event_Order.
+                    // It is not clear where the issue is coming from. The connected event is fired BEFORE the disconnected
+                    // event. This is wrong. It seems that the finally block in HandleClientAsync must be finished before we
+                    // can continue here. Maybe there is a better way to do this.
+                    SpinWait.SpinUntil(() => existingConnection.IsFinalized, TimeSpan.FromSeconds(10));
                 }
 
                 if (isSessionPresent)
@@ -377,13 +390,9 @@ namespace MQTTnet.Server
 
                 return connection;
             }
-            finally
-            {
-                _createConnectionGate.Release();
-            }
         }
 
-        private async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
+        async Task<MqttApplicationMessageInterceptorContext> InterceptApplicationMessageAsync(MqttClientConnection senderConnection, MqttApplicationMessage applicationMessage)
         {
             var interceptor = _options.ApplicationMessageInterceptor;
             if (interceptor == null)
@@ -411,7 +420,7 @@ namespace MQTTnet.Server
             return interceptorContext;
         }
 
-        private async Task TryCleanupChannelAsync(IMqttChannelAdapter channelAdapter)
+        async Task SafeCleanupChannelAsync(IMqttChannelAdapter channelAdapter)
         {
             try
             {

Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs

@@ -1,9 +1,9 @@
-using System;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-using MQTTnet.Packets;
-using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
@@ -67,7 +67,7 @@ namespace MQTTnet.Server
                         _subscriptions[finalTopicFilter.Topic] = finalTopicFilter;
                     }
 
-                    await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
+                    await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
                 }
             }
 
@@ -83,7 +83,7 @@ namespace MQTTnet.Server
                 var interceptorContext = await InterceptSubscribeAsync(topicFilter).ConfigureAwait(false);
                 if (!interceptorContext.AcceptSubscription)
                 {
-                   continue;
+                    continue;
                 }
 
                 if (interceptorContext.AcceptSubscription)
@@ -93,7 +93,7 @@ namespace MQTTnet.Server
                         _subscriptions[topicFilter.Topic] = topicFilter;
                     }
 
-                    await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
+                    await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
                 }
             }
         }
@@ -131,9 +131,9 @@ namespace MQTTnet.Server
 
             foreach (var topicFilter in unsubscribePacket.TopicFilters)
             {
-                await _eventDispatcher.HandleClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
             }
-            
+
             return unsubAckPacket;
         }
 
@@ -152,7 +152,7 @@ namespace MQTTnet.Server
                 lock (_subscriptions)
                 {
                     _subscriptions.Remove(topicFilter);
-                }    
+                }
             }
         }
 

Source/MQTTnet/Server/MqttRetainedMessagesManager.cs

@@ -1,10 +1,10 @@
-using System;
+using MQTTnet.Diagnostics;
+using MQTTnet.Implementations;
+using MQTTnet.Internal;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-using MQTTnet.Diagnostics;
-using MQTTnet.Implementations;
-using MQTTnet.Internal;
 
 namespace MQTTnet.Server
 {
@@ -14,10 +14,10 @@ namespace MQTTnet.Server
         private readonly AsyncLock _messagesLock = new AsyncLock();
         private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();
 
-        private IMqttNetChildLogger _logger;
+        private IMqttNetLogger _logger;
         private IMqttServerOptions _options;
 
-        public Task Start(IMqttServerOptions options, IMqttNetChildLogger logger)
+        public Task Start(IMqttServerOptions options, IMqttNetLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttRetainedMessagesManager));

Source/MQTTnet/Server/MqttServer.cs

@@ -1,15 +1,15 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.Client.Publishing;
 using MQTTnet.Client.Receiving;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Protocol;
 using MQTTnet.Server.Status;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
@@ -17,13 +17,13 @@ namespace MQTTnet.Server
     {
         private readonly MqttServerEventDispatcher _eventDispatcher;
         private readonly ICollection<IMqttServerAdapter> _adapters;
-        private readonly IMqttNetChildLogger _logger;
+        private readonly IMqttNetLogger _logger;
 
         private MqttClientSessionsManager _clientSessionsManager;
         private IMqttRetainedMessagesManager _retainedMessagesManager;
         private CancellationTokenSource _cancellationTokenSource;
 
-        public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetChildLogger logger)
+        public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
         {
             if (adapters == null) throw new ArgumentNullException(nameof(adapters));
             _adapters = adapters.ToList();
@@ -194,7 +194,7 @@ namespace MQTTnet.Server
 
         private Task OnHandleClient(IMqttChannelAdapter channelAdapter)
         {
-            return _clientSessionsManager.HandleClientAsync(channelAdapter);
+            return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter);
         }
     }
 }

Source/MQTTnet/Server/MqttServerEventDispatcher.cs

@@ -1,15 +1,15 @@
-using System;
-using System.Threading.Tasks;
-using MQTTnet.Client.Receiving;
+using MQTTnet.Client.Receiving;
 using MQTTnet.Diagnostics;
+using System;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Server
 {
     public class MqttServerEventDispatcher
     {
-        private readonly IMqttNetChildLogger _logger;
+        readonly IMqttNetLogger _logger;
 
-        public MqttServerEventDispatcher(IMqttNetChildLogger logger)
+        public MqttServerEventDispatcher(IMqttNetLogger logger)
         {
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
@@ -24,18 +24,25 @@ namespace MQTTnet.Server
 
         public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
 
-        public Task HandleClientConnectedAsync(string clientId)
+        public async Task SafeNotifyClientConnectedAsync(string clientId)
         {
-            var handler = ClientConnectedHandler;
-            if (handler == null)
+            try
             {
-                return Task.FromResult(0);
-            }
+                var handler = ClientConnectedHandler;
+                if (handler == null)
+                {
+                    return;
+                }
 
-            return handler.HandleClientConnectedAsync(new MqttServerClientConnectedEventArgs(clientId));
+                await handler.HandleClientConnectedAsync(new MqttServerClientConnectedEventArgs(clientId)).ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error while handling custom 'ClientConnected' event.");
+            }
         }
 
-        public async Task TryHandleClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType)
+        public async Task SafeNotifyClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType)
         {
             try
             {
@@ -49,41 +56,62 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, "Error while handling 'ClientDisconnected' event.");
+                _logger.Error(exception, "Error while handling custom 'ClientDisconnected' event.");
             }
         }
 
-        public Task HandleClientSubscribedTopicAsync(string clientId, TopicFilter topicFilter)
+        public async Task SafeNotifyClientSubscribedTopicAsync(string clientId, TopicFilter topicFilter)
         {
-            var handler = ClientSubscribedTopicHandler;
-            if (handler == null)
+            try
             {
-                return Task.FromResult(0);
-            }
+                var handler = ClientSubscribedTopicHandler;
+                if (handler == null)
+                {
+                    return;
+                }
 
-            return handler.HandleClientSubscribedTopicAsync(new MqttServerClientSubscribedTopicEventArgs(clientId, topicFilter));
+                await handler.HandleClientSubscribedTopicAsync(new MqttServerClientSubscribedTopicEventArgs(clientId, topicFilter)).ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error while handling custom 'ClientSubscribedTopic' event.");
+            }
         }
 
-        public Task HandleClientUnsubscribedTopicAsync(string clientId, string topicFilter)
+        public async Task SafeNotifyClientUnsubscribedTopicAsync(string clientId, string topicFilter)
         {
-            var handler = ClientUnsubscribedTopicHandler;
-            if (handler == null)
+            try
             {
-                return Task.FromResult(0);
-            }
+                var handler = ClientUnsubscribedTopicHandler;
+                if (handler == null)
+                {
+                    return;
+                }
 
-            return handler.HandleClientUnsubscribedTopicAsync(new MqttServerClientUnsubscribedTopicEventArgs(clientId, topicFilter));
+                await handler.HandleClientUnsubscribedTopicAsync(new MqttServerClientUnsubscribedTopicEventArgs(clientId, topicFilter)).ConfigureAwait(false);
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error while handling custom 'ClientUnsubscribedTopic' event.");
+            }
         }
 
-        public Task HandleApplicationMessageReceivedAsync(string senderClientId, MqttApplicationMessage applicationMessage)
+        public async Task SafeNotifyApplicationMessageReceivedAsync(string senderClientId, MqttApplicationMessage applicationMessage)
         {
-            var handler = ApplicationMessageReceivedHandler;
-            if (handler == null)
+            try
             {
-                return Task.FromResult(0);
-            }
+                var handler = ApplicationMessageReceivedHandler;
+                if (handler == null)
+                {
+                    return;
+                }
 
-            return handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage));
+                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage)).ConfigureAwait(false); ;
+            }
+            catch (Exception exception)
+            {
+                _logger.Error(exception, "Error while handling custom 'ApplicationMessageReceived' event.");
+            }
         }
     }
 }

Source/MQTTnet/Server/Status/IMqttClientStatus.cs

@@ -1,6 +1,6 @@
-using System;
+using MQTTnet.Formatter;
+using System;
 using System.Threading.Tasks;
-using MQTTnet.Formatter;
 
 namespace MQTTnet.Server.Status
 {
@@ -9,7 +9,7 @@ namespace MQTTnet.Server.Status
         string ClientId { get; }
 
         string Endpoint { get; }
-        
+
         MqttProtocolVersion ProtocolVersion { get; }
 
         DateTime LastPacketReceivedTimestamp { get; }
@@ -29,7 +29,7 @@ namespace MQTTnet.Server.Status
         long BytesSent { get; }
 
         long BytesReceived { get; }
-        
+
         Task DisconnectAsync();
 
         void ResetStatistics();

Source/MQTTnet/Server/Status/IMqttSessionStatus.cs

@@ -1,12 +1,15 @@
-using System.Threading.Tasks;
+using System.Collections.Generic;
+using System.Threading.Tasks;
 
 namespace MQTTnet.Server.Status
 {
     public interface IMqttSessionStatus
     {
-        string ClientId { get; set; }
+        string ClientId { get; }
 
-        long PendingApplicationMessagesCount { get; set; }
+        long PendingApplicationMessagesCount { get; }
+
+        IDictionary<object, object> Items { get; }
 
         Task ClearPendingApplicationMessagesAsync();
 

Source/MQTTnet/Server/Status/MqttClientStatus.cs

@@ -1,16 +1,16 @@
-using System;
+using MQTTnet.Formatter;
+using System;
 using System.Threading.Tasks;
-using MQTTnet.Formatter;
 
 namespace MQTTnet.Server.Status
 {
     public class MqttClientStatus : IMqttClientStatus
     {
-        private readonly MqttClientConnection _connection;
+        readonly MqttClientConnection _connection;
 
         public MqttClientStatus(MqttClientConnection connection)
         {
-            _connection = connection;
+            _connection = connection ?? throw new ArgumentNullException(nameof(connection));
         }
 
         public string ClientId { get; set; }

Source/MQTTnet/Server/Status/MqttSessionStatus.cs

@@ -1,12 +1,13 @@
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace MQTTnet.Server.Status
 {
     public class MqttSessionStatus : IMqttSessionStatus
     {
-        private readonly MqttClientSession _session;
-        private readonly MqttClientSessionsManager _sessionsManager;
+        readonly MqttClientSession _session;
+        readonly MqttClientSessionsManager _sessionsManager;
 
         public MqttSessionStatus(MqttClientSession session, MqttClientSessionsManager sessionsManager)
         {
@@ -17,14 +18,16 @@ namespace MQTTnet.Server.Status
         public string ClientId { get; set; }
 
         public long PendingApplicationMessagesCount { get; set; }
-    
+
         public DateTime CreatedTimestamp { get; set; }
 
+        public IDictionary<object, object> Items { get; set; }
+
         public Task DeleteAsync()
         {
             return _sessionsManager.DeleteSessionAsync(ClientId);
         }
-        
+
         public Task ClearPendingApplicationMessagesAsync()
         {
             _session.ApplicationMessagesQueue.Clear();

Source/MQTTnet/TopicFilter.cs

@@ -2,6 +2,7 @@
 
 namespace MQTTnet
 {
+    // TODO: Consider renaming to "MqttTopicFilter"
     public class TopicFilter
     {
         public string Topic { get; set; }
@@ -26,16 +27,16 @@ namespace MQTTnet
         public override string ToString()
         {
             return string.Concat(
-                "TopicFilter: [Topic=", 
+                "TopicFilter: [Topic=",
                 Topic,
-                "] [QualityOfServiceLevel=", 
+                "] [QualityOfServiceLevel=",
                 QualityOfServiceLevel,
-                "] [NoLocal=", 
-                NoLocal, 
-                "] [RetainAsPublished=", 
-                RetainAsPublished, 
-                "] [RetainHandling=", 
-                RetainHandling, 
+                "] [NoLocal=",
+                NoLocal,
+                "] [RetainAsPublished=",
+                RetainAsPublished,
+                "] [RetainHandling=",
+                RetainHandling,
                 "]");
         }
     }