Add Unit Tests

Fixes https://github.com/dotnet/MQTTnet/issues/1765.

.github/workflows/ReleaseNotes.md

@@ -1,4 +1,4 @@
-* [Core] Add validation of maximum string lengths (#1718).
-* [Client] Added overloads for setting packet payload and will payload (#1720).
-* [Client] The proper connect result is now exposed in the _Disconnected_ event when authentication fails (#1139). 
-* [Server] Improved performance by changing internal locking strategy for subscriptions (#1716, thanks to @zeheng).
+* [Client] Fixed _PlatformNotSupportedException_ when using Blazor (#1755, thanks to @Nickztar).
+* [Server] Fixed _NullReferenceException_ in retained messages management (#1762, thanks to @logicaloud).
+* [Server] Exposed new option which allows disabling packet fragmentation (#1753).
+* [Server] Expired sessions will no longer be used when a client connects (#1756).

Samples/Client/Client_Subscribe_Samples.cs

@@ -6,6 +6,7 @@
 // ReSharper disable UnusedMember.Global
 // ReSharper disable InconsistentNaming
 
+using MQTTnet;
 using MQTTnet.Client;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
@@ -176,4 +177,4 @@ public static class Client_Subscribe_Samples
             response.DumpToConsole();
         }
     }
-}
+}

Samples/Server/Server_Retained_Messages_Samples.cs

@@ -7,6 +7,8 @@
 // ReSharper disable InconsistentNaming
 
 using System.Text.Json;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.Samples.Server;
 
@@ -32,7 +34,10 @@ public static class Server_Retained_Messages_Samples
             {
                 try
                 {
-                    eventArgs.LoadedRetainedMessages = await JsonSerializer.DeserializeAsync<List<MqttApplicationMessage>>(File.OpenRead(storePath));
+                    var models = await JsonSerializer.DeserializeAsync<List<MqttRetainedMessageModel>>(File.OpenRead(storePath)) ?? new List<MqttRetainedMessageModel>();
+                    var retainedMessages = models.Select(m => m.ToApplicationMessage()).ToList();
+
+                    eventArgs.LoadedRetainedMessages = retainedMessages;
                     Console.WriteLine("Retained messages loaded.");
                 }
                 catch (FileNotFoundException)
@@ -56,7 +61,9 @@ public static class Server_Retained_Messages_Samples
                     // used to write all retained messages to dedicated files etc. Then all files must be loaded and a full list
                     // of retained messages must be provided in the loaded event.
 
-                    var buffer = JsonSerializer.SerializeToUtf8Bytes(eventArgs.StoredRetainedMessages);
+                    var models = eventArgs.StoredRetainedMessages.Select(MqttRetainedMessageModel.Create);
+
+                    var buffer = JsonSerializer.SerializeToUtf8Bytes(models);
                     await File.WriteAllBytesAsync(storePath, buffer);
                     Console.WriteLine("Retained messages saved.");
                 }
@@ -79,4 +86,59 @@ public static class Server_Retained_Messages_Samples
             Console.ReadLine();
         }
     }
+
+    sealed class MqttRetainedMessageModel
+    {
+        public string? ContentType { get; set; }
+        public byte[]? CorrelationData { get; set; }
+        public byte[]? Payload { get; set; }
+        public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; }
+        public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; }
+        public string? ResponseTopic { get; set; }
+        public string? Topic { get; set; }
+        public List<MqttUserProperty>? UserProperties { get; set; }
+
+        public static MqttRetainedMessageModel Create(MqttApplicationMessage message)
+        {
+            if (message == null)
+            {
+                throw new ArgumentNullException(nameof(message));
+            }
+
+            return new MqttRetainedMessageModel
+            {
+                Topic = message.Topic,
+
+                // Create a copy of the buffer from the payload segment because 
+                // it cannot be serialized and deserialized with the JSON serializer.
+                Payload = message.PayloadSegment.ToArray(),
+                UserProperties = message.UserProperties,
+                ResponseTopic = message.ResponseTopic,
+                CorrelationData = message.CorrelationData,
+                ContentType = message.ContentType,
+                PayloadFormatIndicator = message.PayloadFormatIndicator,
+                QualityOfServiceLevel = message.QualityOfServiceLevel
+
+                // Other properties like "Retain" are not if interest in the storage.
+                // That's why a custom model makes sense.
+            };
+        }
+
+        public MqttApplicationMessage ToApplicationMessage()
+        {
+            return new MqttApplicationMessage
+            {
+                Topic = Topic,
+                PayloadSegment = new ArraySegment<byte>(Payload ?? Array.Empty<byte>()),
+                PayloadFormatIndicator = PayloadFormatIndicator,
+                ResponseTopic = ResponseTopic,
+                CorrelationData = CorrelationData,
+                ContentType = ContentType,
+                UserProperties = UserProperties,
+                QualityOfServiceLevel = QualityOfServiceLevel,
+                Dup = false,
+                Retain = true
+            };
+        }
+    }
 }

Source/MQTTnet.Extensions.ManagedClient/MQTTnet.Extensions.ManagedClient.csproj

@@ -2,7 +2,7 @@
 
     <PropertyGroup>
         <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
-        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
         <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
         
         <AssemblyName>MQTTnet.Extensions.ManagedClient</AssemblyName>

Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj

@@ -2,7 +2,7 @@
 
     <PropertyGroup>
         <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
-        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
         <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
         
         <AssemblyName>MQTTnet.Extensions.Rpc</AssemblyName>

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -69,6 +69,11 @@ namespace MQTTnet.Extensions.Rpc
                 throw new ArgumentNullException(nameof(methodName));
             }
 
+            if (!_mqttClient.IsConnected)
+            {
+                throw new MqttClientDisconnectedException(null);
+            }
+
             var context = new TopicGenerationContext(_mqttClient, _options, methodName, qualityOfServiceLevel);
             var topicNames = _options.TopicGenerationStrategy.CreateRpcTopics(context);
 

Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj

@@ -2,7 +2,7 @@
 
     <PropertyGroup>
         <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
-        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
         <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
         
         <AssemblyName>MQTTnet.Extensions.WebSocket4Net</AssemblyName>

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs

@@ -3,13 +3,16 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
 using MQTTnet.Exceptions;
+using MQTTnet.Extensions.Rpc;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
+using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 
@@ -18,6 +21,38 @@ namespace MQTTnet.Tests.Clients.MqttClient
     [TestClass]
     public sealed class MqttClient_Connection_Tests : BaseTestClass
     {
+        [TestMethod]
+        public async Task Connect_Disconnect_Connect()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var clientOptions = testEnvironment.CreateDefaultClientOptions();
+                var client = testEnvironment.CreateClient();
+
+                await client.ConnectAsync(clientOptions);
+                await client.DisconnectAsync();
+                await client.ConnectAsync(clientOptions);
+            }
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(InvalidOperationException))]
+        public async Task Connect_Multiple_Times_Should_Fail()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var clientOptions = testEnvironment.CreateDefaultClientOptions();
+                var client = testEnvironment.CreateClient();
+
+                await client.ConnectAsync(clientOptions);
+                await client.ConnectAsync(clientOptions);
+            }
+        }
+
         [TestMethod]
         [ExpectedException(typeof(MqttCommunicationException))]
         public async Task Connect_To_Invalid_Server_Port_Not_Opened()
@@ -128,7 +163,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
                 await client.DisconnectAsync(disconnectOptions);
 
                 await LongTestDelay();
-                
+
                 Assert.IsNotNull(eventArgs);
                 Assert.AreEqual(MqttDisconnectReasonCode.MessageRateTooHigh, eventArgs.ReasonCode);
             }
@@ -156,7 +191,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
                 await client.DisconnectAsync(disconnectOptions);
 
                 await LongTestDelay();
-                
+
                 Assert.IsNotNull(eventArgs);
                 Assert.IsNotNull(eventArgs.UserProperties);
                 Assert.AreEqual(1, eventArgs.UserProperties.Count);
@@ -164,5 +199,47 @@ namespace MQTTnet.Tests.Clients.MqttClient
                 Assert.AreEqual("test_value", eventArgs.UserProperties[0].Value);
             }
         }
+
+        [TestMethod]
+        public async Task Return_Non_Success()
+        {
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
+            {
+                var server = await testEnvironment.StartServer();
+
+                server.ValidatingConnectionAsync += args =>
+                {
+                    args.ResponseUserProperties = new List<MqttUserProperty>
+                    {
+                        new MqttUserProperty("Property", "Value")
+                    };
+
+                    args.ReasonCode = MqttConnectReasonCode.QuotaExceeded;
+
+                    return CompletedTask.Instance;
+                };
+
+                var client = testEnvironment.CreateClient();
+
+                var response = await client.ConnectAsync(testEnvironment.CreateDefaultClientOptionsBuilder().WithoutThrowOnNonSuccessfulConnectResponse().Build());
+
+                Assert.IsNotNull(response);
+                Assert.AreEqual(MqttClientConnectResultCode.QuotaExceeded, response.ResultCode);
+                Assert.AreEqual(response.UserProperties[0].Name, "Property");
+                Assert.AreEqual(response.UserProperties[0].Value, "Value");
+            }
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(MqttClientDisconnectedException))]
+        public async Task Rpc_Client_Not_Connected()
+        {
+            var factory = new MqttFactory();
+            using (var client = factory.CreateMqttClient())
+            {
+                var rpcClient = factory.CreateMqttRpcClient(client);
+                await rpcClient.ExecuteAsync("aaa", new byte[] { 1, 2, 3 }, MqttQualityOfServiceLevel.AtLeastOnce);
+            }
+        }
     }
 }

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Publish_Tests.cs

@@ -0,0 +1,59 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Protocol;
+using MQTTnet.Server;
+
+namespace MQTTnet.Tests.Clients.MqttClient
+{
+    [TestClass]
+    public sealed class MqttClient_Publish_Tests : BaseTestClass
+    {
+        [TestMethod]
+        public async Task Acknowledge_After_Disconnect_Should_Fail()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                var client = await testEnvironment.ConnectClient();
+                await client.SubscribeAsync("TEST", MqttQualityOfServiceLevel.AtLeastOnce);
+
+                var flag = false;
+                
+                client.ApplicationMessageReceivedAsync += async args =>
+                {
+                    try
+                    {
+                        args.AutoAcknowledge = false;
+                        
+                        await client.DisconnectAsync();
+                        await args.AcknowledgeAsync();
+
+                        flag = true;
+                    }
+                    catch (Exception exception)
+                    {
+                        
+                    }
+                };
+
+                await server.InjectApplicationMessageAsync(
+                    new MqttApplicationMessage
+                    {
+                        Topic = "TEST"
+                    });
+
+                SpinWait.SpinUntil(() => flag, TimeSpan.FromSeconds(10));
+                
+                Assert.IsTrue(flag);
+            }
+        }
+    }
+}

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs

@@ -69,38 +69,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
                 Assert.IsTrue(success);
             }
         }
+       
 
-        [TestMethod]
-        public async Task Connect_Disconnect_Connect()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                await testEnvironment.StartServer();
-
-                var clientOptions = testEnvironment.CreateDefaultClientOptions();
-                var client = testEnvironment.CreateClient();
-
-                await client.ConnectAsync(clientOptions);
-                await client.DisconnectAsync();
-                await client.ConnectAsync(clientOptions);
-            }
-        }
-        
-        [TestMethod]
-        [ExpectedException(typeof(InvalidOperationException))]
-        public async Task Connect_Multiple_Times_Should_Fail()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                await testEnvironment.StartServer();
-
-                var clientOptions = testEnvironment.CreateDefaultClientOptions();
-                var client = testEnvironment.CreateClient();
-
-                await client.ConnectAsync(clientOptions);
-                await client.ConnectAsync(clientOptions);
-            }
-        }
 
         [TestMethod]
         public async Task Disconnect_Event_Contains_Exception()

Source/MQTTnet.Tests/Clients/MqttClientOptionsBuilder_Tests.cs

@@ -10,7 +10,7 @@ using MQTTnet.Client;
 namespace MQTTnet.Tests.Clients
 {
     [TestClass]
-    public class MqttClientOptionsBuilder_Tests
+    public sealed class MqttClientOptionsBuilder_Tests
     {
         [TestMethod]
         public void WithConnectionUri_Credential_Test()

Source/MQTTnet.Tests/Extension_Tests.cs

@@ -1,98 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.VisualStudio.TestTools.UnitTesting;
-using MQTTnet.Exceptions;
-using MQTTnet.Internal;
-
-namespace MQTTnet.Tests
-{
-    [TestClass]
-    public class Extension_Tests
-    {
-        [ExpectedException(typeof(MqttCommunicationTimedOutException))]
-        [TestMethod]
-        public async Task TimeoutAfter()
-        {
-            await MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct),
-                TimeSpan.FromMilliseconds(100), CancellationToken.None);
-        }
-
-        [ExpectedException(typeof(MqttCommunicationTimedOutException))]
-        [TestMethod]
-        public async Task TimeoutAfterWithResult()
-        {
-            await MqttTaskTimeout.WaitAsync(
-                ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct).ContinueWith(t => 5, ct),
-                TimeSpan.FromMilliseconds(100), CancellationToken.None);
-        }
-        
-        [TestMethod]
-        public async Task TimeoutAfterWithInnerException()
-        {
-            try
-            {
-                await MqttTaskTimeout.WaitAsync(ct => Task.Run(() =>
-                {
-                    var iis = new int[0];
-                    iis[1] = 0;
-                }, ct), TimeSpan.FromSeconds(1), CancellationToken.None);
-
-                Assert.Fail();
-            }
-            catch (Exception e)
-            {
-                Assert.IsTrue(e is IndexOutOfRangeException);
-            }
-        }
-
-        [TestMethod]
-        public async Task TimeoutAfterWithInnerExceptionWithResult()
-        {
-            try
-            {
-                await MqttTaskTimeout.WaitAsync(ct => Task.Run(() =>
-                {
-                    var iis = new int[0];
-                    iis[1] = 0;
-                    return iis[0];
-                }, ct), TimeSpan.FromSeconds(1), CancellationToken.None);
-
-                Assert.Fail();
-            }
-            catch (Exception e)
-            {
-                Assert.IsTrue(e is IndexOutOfRangeException);
-            }
-        }
-
-        [TestMethod]
-        public async Task TimeoutAfterMemoryUsage()
-        {
-            var initialMemory = GC.GetTotalMemory(true);
-
-            var tasks = Enumerable.Range(0, 100000)
-                .Select(i =>
-                {
-                    return MqttTaskTimeout.WaitAsync(ct => Task.Delay(TimeSpan.FromMilliseconds(1), ct),
-                        TimeSpan.FromMinutes(1), CancellationToken.None);
-                });
-
-            await Task.WhenAll(tasks);
-            AssertIsLess(initialMemory + 2_000_000, GC.GetTotalMemory(true));
-        }
-
-        static void AssertIsLess(long bound, long actual)
-        {
-            if (bound < actual)
-            {
-                Assert.Fail($"value must be less than {bound:N0} but is {actual:N0}");
-            }
-        }
-    }
-}

Source/MQTTnet.Tests/Formatter/MqttPacketSerialization_V3_Tests.cs

@@ -78,7 +78,7 @@ namespace MQTTnet.Tests.Formatter
             Assert.IsNull(deserialized.UserProperties); // Not supported in v3.1.1
         }
         
-                [TestMethod]
+        [TestMethod]
         public void Serialize_Full_MqttConnAckPacket_V310()
         {
             var connAckPacket = new MqttConnAckPacket

Source/MQTTnet.Tests/Mockups/TestEnvironment.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
@@ -24,10 +25,9 @@ namespace MQTTnet.Tests.Mockups
     public sealed class TestEnvironment : IDisposable
     {
         readonly List<string> _clientErrors = new List<string>();
-        readonly List<IMqttClient> _clients = new List<IMqttClient>();
-
+        readonly ConcurrentBag<IMqttClient> _clients = new ConcurrentBag<IMqttClient>();
         readonly List<Exception> _exceptions = new List<Exception>();
-        readonly List<ILowLevelMqttClient> _lowLevelClients = new List<ILowLevelMqttClient>();
+        readonly ConcurrentBag<ILowLevelMqttClient> _lowLevelClients = new ConcurrentBag<ILowLevelMqttClient>();
         readonly MqttProtocolVersion _protocolVersion;
         readonly List<string> _serverErrors = new List<string>();
 
@@ -65,9 +65,12 @@ namespace MQTTnet.Tests.Mockups
 
                 if (e.LogMessage.Level == MqttNetLogLevel.Error)
                 {
-                    lock (_clientErrors)
+                    if (!IgnoreClientLogErrors)
                     {
-                        _clientErrors.Add(e.LogMessage.ToString());
+                        lock (_clientErrors)
+                        {
+                            _clientErrors.Add(e.LogMessage.ToString());
+                        }
                     }
                 }
             };
@@ -212,30 +215,27 @@ namespace MQTTnet.Tests.Mockups
 
         public IMqttClient CreateClient()
         {
-            lock (_clients)
+            var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : MqttNetNullLogger.Instance;
+            
+            var client = Factory.CreateMqttClient(logger);
+            client.ConnectingAsync += e =>
             {
-                var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : new MqttNetNullLogger();
-
-                var client = Factory.CreateMqttClient(logger);
-                _clients.Add(client);
-
-                client.ConnectingAsync += e =>
+                if (TestContext != null)
                 {
-                    if (TestContext != null)
+                    var clientOptions = e.ClientOptions;
+                    var existingClientId = clientOptions.ClientId;
+                    if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
                     {
-                        var clientOptions = e.ClientOptions;
-                        var existingClientId = clientOptions.ClientId;
-                        if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
-                        {
-                            clientOptions.ClientId = TestContext.TestName + "_" + existingClientId;
-                        }
+                        clientOptions.ClientId = TestContext.TestName + "_" + existingClientId;
                     }
+                }
 
-                    return CompletedTask.Instance;
-                };
-
-                return client;
-            }
+                return CompletedTask.Instance;
+            };
+            
+            _clients.Add(client);
+            
+            return client;
         }
 
         public MqttClientOptions CreateDefaultClientOptions()
@@ -250,13 +250,10 @@ namespace MQTTnet.Tests.Mockups
 
         public ILowLevelMqttClient CreateLowLevelClient()
         {
-            lock (_clients)
-            {
-                var client = Factory.CreateLowLevelMqttClient(ClientLogger);
-                _lowLevelClients.Add(client);
+            var client = Factory.CreateLowLevelMqttClient(ClientLogger);
+            _lowLevelClients.Add(client);
 
-                return client;
-            }
+            return client;
         }
 
         public MqttServer CreateServer(MqttServerOptions options)
@@ -383,13 +380,16 @@ namespace MQTTnet.Tests.Mockups
                 }
             }
 
-            lock (_clientErrors)
+            if (!IgnoreClientLogErrors)
             {
-                if (!IgnoreClientLogErrors && _clientErrors.Count > 0)
+                lock (_clientErrors)
                 {
-                    var message = $"Client(s) had {_clientErrors.Count} errors (${string.Join(Environment.NewLine, _clientErrors)})";
-                    Console.WriteLine(message);
-                    throw new Exception(message);
+                    if (_clientErrors.Count > 0)
+                    {
+                        var message = $"Client(s) had {_clientErrors.Count} errors (${string.Join(Environment.NewLine, _clientErrors)})";
+                        Console.WriteLine(message);
+                        throw new Exception(message);
+                    }
                 }
             }
         }

Source/MQTTnet.Tests/Server/General.cs

@@ -598,8 +598,7 @@ namespace MQTTnet.Tests.Server
                 // c
                 var c1 = await testEnvironment.ConnectClient(clientOptionsBuilder);
 
-                await Task.Delay(500);
-
+                await LongTestDelay();
                 var flow = string.Join(string.Empty, events);
                 Assert.AreEqual("c", flow);
 
@@ -607,10 +606,8 @@ namespace MQTTnet.Tests.Server
                 // Connect client with same client ID. Should disconnect existing client.
                 var c2 = await testEnvironment.ConnectClient(clientOptionsBuilder);
 
-                await Task.Delay(500);
-
+                await LongTestDelay();
                 flow = string.Join(string.Empty, events);
-
                 Assert.AreEqual("cdc", flow);
 
                 c2.ApplicationMessageReceivedAsync += e =>
@@ -628,8 +625,7 @@ namespace MQTTnet.Tests.Server
                 // r
                 await c2.PublishStringAsync("topic");
 
-                await Task.Delay(500);
-
+                await LongTestDelay();
                 flow = string.Join(string.Empty, events);
                 Assert.AreEqual("cdcr", flow);
 
@@ -639,16 +635,17 @@ namespace MQTTnet.Tests.Server
                 await c1.TryDisconnectAsync();
                 Assert.AreEqual(false, c1.IsConnected);
 
-                await Task.Delay(500);
+                await LongTestDelay();
 
                 // d
                 Assert.AreEqual(true, c2.IsConnected);
                 await c2.DisconnectAsync();
 
-                await Task.Delay(500);
+                await LongTestDelay();
 
                 await server.StopAsync();
 
+                await LongTestDelay();
                 flow = string.Join(string.Empty, events);
                 Assert.AreEqual("cdcrd", flow);
             }
@@ -852,11 +849,11 @@ namespace MQTTnet.Tests.Server
                     return CompletedTask.Instance;
                 };
 
-                await Task.Delay(100);
-
+                // Stopping the server should disconnect the connection with the client
+                // which, it turn, will fire the disconnected event at the client.
                 await server.StopAsync();
 
-                await Task.Delay(100);
+                await LongTestDelay();
 
                 Assert.AreEqual(1, disconnectCalled);
             }

Source/MQTTnet.Tests/Server/MqttRetainedMessageManager_Tests.cs

@@ -0,0 +1,25 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Tests.Server
+{
+    [TestClass]
+    public sealed class MqttRetainedMessageManager_Tests
+    {
+        [TestMethod]
+        public async Task MqttRetainedMessageManager_GetUndefinedTopic()
+        {
+            var logger = new Mockups.TestLogger();
+            var eventContainer = new MQTTnet.Server.MqttServerEventContainer();
+            var retainedMessagesManager = new MQTTnet.Server.MqttRetainedMessagesManager(eventContainer, logger);
+            var task = retainedMessagesManager.GetMessage("undefined");
+            Assert.IsNotNull(task, "Task should not be null");
+            var result = await task;
+            Assert.IsNull(result, "Null result expected");
+        }
+    }
+}

Source/MQTTnet.Tests/Server/MqttSubscriptionsManager_Tests.cs

@@ -221,7 +221,10 @@ namespace MQTTnet.Tests.Server
             var eventContainer = new MqttServerEventContainer();
             var clientSessionManager = new MqttClientSessionsManager(options, retainedMessagesManager, eventContainer, logger);
 
-            var session = new MqttSession("", false, new ConcurrentDictionary<object, object>(), options, eventContainer, retainedMessagesManager, clientSessionManager);
+            var session = new MqttSession(new MqttConnectPacket
+            {
+                ClientId = ""
+            }, new ConcurrentDictionary<object, object>(), options, eventContainer, retainedMessagesManager, clientSessionManager);
 
             _subscriptionsManager = new MqttClientSubscriptionsManager(session, new MqttServerEventContainer(), retainedMessagesManager, clientSessionManager);
         }

Source/MQTTnet.Tests/Server/Session_Tests.cs

@@ -70,6 +70,28 @@ namespace MQTTnet.Tests.Server
             }
         }
 
+        [TestMethod]
+        public async Task Do_Not_Use_Expired_Session()
+        {
+            using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
+            {
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
+
+                var c1 = await testEnvironment.ConnectClient(o => o.WithClientId("Client1").WithCleanSession(false).WithSessionExpiryInterval(3));
+
+                // Kill the client connection and ensure that the session will stay there but is expired after 3 seconds!
+                c1.Dispose();
+
+                await Task.Delay(TimeSpan.FromSeconds(6));
+
+                c1 = testEnvironment.CreateClient();
+                var options = testEnvironment.CreateDefaultClientOptionsBuilder().WithClientId("Client1").WithCleanSession(false).WithSessionExpiryInterval(3).Build();
+
+                var connectResult = await c1.ConnectAsync(options);
+                Assert.AreEqual(false, connectResult.IsSessionPresent);
+            }
+        }
+
         [TestMethod]
         public async Task Fire_Deleted_Event()
         {
@@ -121,7 +143,7 @@ namespace MQTTnet.Tests.Server
                 Assert.AreEqual(true, session.Items["can_subscribe_x"]);
             }
         }
-        
+
         [TestMethod]
         [DataRow(MqttProtocolVersion.V310)]
         [DataRow(MqttProtocolVersion.V311)]
@@ -134,16 +156,12 @@ namespace MQTTnet.Tests.Server
 
                 await testEnvironment.StartServer();
 
-                var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(1)).WithProtocolVersion(protocolVersion);
+                var options = new MqttClientOptionsBuilder().WithClientId("1").WithTimeout(TimeSpan.FromSeconds(10)).WithProtocolVersion(protocolVersion);
 
                 var hasReceive = false;
-
                 void OnReceive()
                 {
-                    if (!hasReceive)
-                    {
-                        hasReceive = true;
-                    }
+                    hasReceive = true;
                 }
 
                 // Try to connect 50 clients at the same time.
@@ -152,7 +170,7 @@ namespace MQTTnet.Tests.Server
                 var connectedClients = clients.Where(c => c?.TryPingAsync().GetAwaiter().GetResult() == true).ToList();
 
                 await LongTestDelay();
-                
+
                 Assert.AreEqual(1, connectedClients.Count);
 
                 var option2 = new MqttClientOptionsBuilder().WithClientId("2").WithKeepAlivePeriod(TimeSpan.FromSeconds(10));
@@ -164,7 +182,7 @@ namespace MQTTnet.Tests.Server
                 Assert.AreEqual(true, hasReceive);
             }
         }
- 
+
         [DataTestMethod]
         [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
         [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
@@ -355,8 +373,8 @@ namespace MQTTnet.Tests.Server
                     onReceive();
                     return CompletedTask.Instance;
                 };
-                
-                using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)))
+
+                using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30)))
                 {
                     await client.SubscribeAsync("aaa", MqttQualityOfServiceLevel.AtMostOnce, timeout.Token).ConfigureAwait(false);
                 }

Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs

@@ -557,7 +557,7 @@ namespace MQTTnet.Tests.Server
             var eventContainer = new MqttServerEventContainer();
             var retainedMessagesManager = new MqttRetainedMessagesManager(eventContainer, logger);
             var sessionManager = new MqttClientSessionsManager(serverOptions, retainedMessagesManager, eventContainer, logger);
-            _clientSession = new MqttSession(clientId, false, new Dictionary<object, object>(), serverOptions, eventContainer, retainedMessagesManager, sessionManager);
+            _clientSession = new MqttSession(new MqttConnectPacket{ ClientId = clientId }, new Dictionary<object, object>(), serverOptions, eventContainer, retainedMessagesManager, sessionManager);
 
             List<string> topicsToSubscribe;
 

Source/MQTTnet/Client/Internal/MqttClientResultFactory.cs

@@ -6,6 +6,7 @@ namespace MQTTnet.Client.Internal
 {
     public static class MqttClientResultFactory
     {
+        public static readonly MqttClientConnectResultFactory ConnectResult = new MqttClientConnectResultFactory();
         public static readonly MqttClientPublishResultFactory PublishResult = new MqttClientPublishResultFactory();
         public static readonly MqttClientSubscribeResultFactory SubscribeResult = new MqttClientSubscribeResultFactory();
         public static readonly MqttClientUnsubscribeResultFactory UnsubscribeResult = new MqttClientUnsubscribeResultFactory();

Source/MQTTnet/Client/MqttClient.cs

@@ -121,11 +121,11 @@ namespace MQTTnet.Client
                 var mqttClientAliveToken = _mqttClientAlive.Token;
 
                 var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_events.InspectPacketEvent, _rootLogger), _rootLogger);
-                _adapter = adapter;
+                _adapter = adapter ?? throw new InvalidOperationException("The adapter factory did not provide an adapter.");
 
                 if (cancellationToken.CanBeCanceled)
                 {
-                    connectResult = await ConnectInternal(cancellationToken).ConfigureAwait(false);
+                    connectResult = await ConnectInternal(adapter, cancellationToken).ConfigureAwait(false);
                 }
                 else
                 {
@@ -133,10 +133,16 @@ namespace MQTTnet.Client
                     // CancellationToken.None or similar.
                     using (var timeout = new CancellationTokenSource(Options.Timeout))
                     {
-                        connectResult = await ConnectInternal(timeout.Token).ConfigureAwait(false);
+                        connectResult = await ConnectInternal(adapter, timeout.Token).ConfigureAwait(false);
                     }
                 }
 
+                if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
+                {
+                    _logger.Warning("Connecting failed: {0}", connectResult.ResultCode);
+                    return connectResult;
+                }
+
                 _lastPacketSentTimestamp = DateTime.UtcNow;
 
                 var keepAliveInterval = Options.KeepAlivePeriod;
@@ -421,7 +427,7 @@ namespace MQTTnet.Client
             return CompletedTask.Instance;
         }
 
-        async Task<MqttClientConnectResult> Authenticate(MqttClientOptions options, CancellationToken cancellationToken)
+        async Task<MqttClientConnectResult> Authenticate(IMqttChannelAdapter channelAdapter, MqttClientOptions options, CancellationToken cancellationToken)
         {
             MqttClientConnectResult result;
 
@@ -434,12 +440,19 @@ namespace MQTTnet.Client
 
                 if (receivedPacket is MqttConnAckPacket connAckPacket)
                 {
-                    var clientConnectResultFactory = new MqttClientConnectResultFactory();
-                    result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
+                    result = MqttClientResultFactory.ConnectResult.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
+                }
+                else if (receivedPacket is MqttAuthPacket)
+                {
+                    throw new NotSupportedException("Extended authentication handler is not yet supported");
+                }
+                else if (receivedPacket == null)
+                {
+                    throw new MqttCommunicationException("Connection closed.");
                 }
                 else
                 {
-                    throw new NotSupportedException("Extended authentication handler is not yet supported");
+                    throw new InvalidOperationException($"Received an unexpected MQTT packet ({receivedPacket}).");
                 }
             }
             catch (Exception exception)
@@ -447,9 +460,18 @@ namespace MQTTnet.Client
                 throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null);
             }
 
-            if (result.ResultCode != MqttClientConnectResultCode.Success)
+            // This is no feature. It is basically a backward compatibility option and should be removed in the future.
+            // The client should not throw any exception if the transport layer connection was successful and the server
+            // did send a proper ACK packet with a non success response.
+            if (options.ThrowOnNonSuccessfulConnectResponse)
             {
-                throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
+                _logger.Warning(
+                    "Client will now throw an _MqttConnectingFailedException_. This is obsolete and will be removed in the future. Consider setting _ThrowOnNonSuccessfulResponseFromServer=False_ in client options.");
+
+                if (result.ResultCode != MqttClientConnectResultCode.Success)
+                {
+                    throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
+                }
             }
 
             _logger.Verbose("Authenticated MQTT connection with server established.");
@@ -484,7 +506,7 @@ namespace MQTTnet.Client
             return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand);
         }
 
-        async Task<MqttClientConnectResult> ConnectInternal(CancellationToken cancellationToken)
+        async Task<MqttClientConnectResult> ConnectInternal(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
         {
             var backgroundCancellationToken = _mqttClientAlive.Token;
 
@@ -497,10 +519,12 @@ namespace MQTTnet.Client
                 _publishPacketReceiverQueue?.Dispose();
                 _publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
 
-                var connectResult = await Authenticate(Options, effectiveCancellationToken.Token).ConfigureAwait(false);
-
-                _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken);
-                _packetReceiverTask = Task.Run(() => ReceivePacketsLoop(backgroundCancellationToken), backgroundCancellationToken);
+                var connectResult = await Authenticate(channelAdapter, Options, effectiveCancellationToken.Token).ConfigureAwait(false);
+                if (connectResult.ResultCode == MqttClientConnectResultCode.Success)
+                {
+                    _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken);
+                    _packetReceiverTask = Task.Run(() => ReceivePacketsLoop(backgroundCancellationToken), backgroundCancellationToken);
+                }
 
                 return connectResult;
             }
@@ -531,7 +555,7 @@ namespace MQTTnet.Client
 
             try
             {
-                _packetDispatcher.Dispose(new MqttClientDisconnectedException(exception));
+                _packetDispatcher?.Dispose(new MqttClientDisconnectedException(exception));
 
                 var receiverTask = _packetReceiverTask.WaitAsync(sender, _logger);
                 var publishPacketReceiverTask = _publishPacketReceiverTask.WaitAsync(sender, _logger);
@@ -754,6 +778,65 @@ namespace MQTTnet.Client
             return packet;
         }
 
+        async Task ReceivePacketsLoop(CancellationToken cancellationToken)
+        {
+            try
+            {
+                _logger.Verbose("Start receiving packets.");
+
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    var packet = await Receive(cancellationToken).ConfigureAwait(false);
+
+                    if (cancellationToken.IsCancellationRequested)
+                    {
+                        return;
+                    }
+
+                    if (packet == null)
+                    {
+                        await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false);
+
+                        return;
+                    }
+
+                    await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            catch (Exception exception)
+            {
+                if (_cleanDisconnectInitiated)
+                {
+                    return;
+                }
+
+                if (exception is AggregateException aggregateException)
+                {
+                    exception = aggregateException.GetBaseException();
+                }
+
+                if (exception is OperationCanceledException)
+                {
+                }
+                else if (exception is MqttCommunicationException)
+                {
+                    _logger.Warning(exception, "Communication error while receiving packets.");
+                }
+                else
+                {
+                    _logger.Error(exception, "Error while receiving packets.");
+                }
+
+                _packetDispatcher.FailAll(exception);
+
+                await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
+            }
+            finally
+            {
+                _logger.Verbose("Stopped receiving packets.");
+            }
+        }
+
         async Task<TResponsePacket> Request<TResponsePacket>(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket
         {
             cancellationToken.ThrowIfCancellationRequested();
@@ -798,6 +881,7 @@ namespace MQTTnet.Client
 
             _lastPacketSentTimestamp = DateTime.UtcNow;
 
+            //ThrowIfNotConnected();
             return _adapter.SendPacketAsync(packet, cancellationToken);
         }
 
@@ -920,65 +1004,6 @@ namespace MQTTnet.Client
             }
         }
 
-        async Task ReceivePacketsLoop(CancellationToken cancellationToken)
-        {
-            try
-            {
-                _logger.Verbose("Start receiving packets.");
-
-                while (!cancellationToken.IsCancellationRequested)
-                {
-                    var packet = await Receive(cancellationToken).ConfigureAwait(false);
-
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return;
-                    }
-
-                    if (packet == null)
-                    {
-                        await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false);
-
-                        return;
-                    }
-
-                    await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false);
-                }
-            }
-            catch (Exception exception)
-            {
-                if (_cleanDisconnectInitiated)
-                {
-                    return;
-                }
-
-                if (exception is AggregateException aggregateException)
-                {
-                    exception = aggregateException.GetBaseException();
-                }
-
-                if (exception is OperationCanceledException)
-                {
-                }
-                else if (exception is MqttCommunicationException)
-                {
-                    _logger.Warning(exception, "Communication error while receiving packets.");
-                }
-                else
-                {
-                    _logger.Error(exception, "Error while receiving packets.");
-                }
-
-                _packetDispatcher.FailAll(exception);
-
-                await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
-            }
-            finally
-            {
-                _logger.Verbose("Stopped receiving packets.");
-            }
-        }
-
         async Task TrySendKeepAliveMessages(CancellationToken cancellationToken)
         {
             try

Source/MQTTnet/Client/Options/MqttClientOptions.cs

@@ -12,6 +12,14 @@ namespace MQTTnet.Client
 {
     public sealed class MqttClientOptions
     {
+        /// <summary>
+        ///     Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
+        ///     or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and
+        ///     will close the connection when receiving such packets. If such a service is used this flag must
+        ///     be set to _false_.
+        /// </summary>
+        public bool AllowPacketFragmentation { get; set; } = true;
+
         /// <summary>
         ///     Gets or sets the authentication data.
         ///     <remarks>MQTT 5.0.0+ feature.</remarks>
@@ -24,14 +32,6 @@ namespace MQTTnet.Client
         /// </summary>
         public string AuthenticationMethod { get; set; }
 
-        /// <summary>
-        ///     Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
-        ///     or WebSocket frames etc. Unfortunately not all brokers (like Amazon Web Services (AWS)) do support this feature and
-        ///     will close the connection when receiving such packets. If such a service is used this flag must
-        ///     be set to _false_.
-        /// </summary>
-        public bool AllowPacketFragmentation { get; set; } = true;
-
         public IMqttClientChannelOptions ChannelOptions { get; set; }
 
         /// <summary>
@@ -100,6 +100,11 @@ namespace MQTTnet.Client
         /// </summary>
         public uint SessionExpiryInterval { get; set; }
 
+        /// <summary>
+        ///     Gets or sets whether an exception should be thrown when the server has sent a non success ACK packet.
+        /// </summary>
+        public bool ThrowOnNonSuccessfulConnectResponse { get; set; } = true;
+
         /// <summary>
         ///     Gets or sets the timeout which will be applied at socket level and internal operations.
         ///     The default value is the same as for sockets in .NET in general.
@@ -222,4 +227,4 @@ namespace MQTTnet.Client
         /// </summary>
         public int WriterBufferSizeMax { get; set; } = 65535;
     }
-}
+}

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -78,7 +78,17 @@ namespace MQTTnet.Client
 
             return _options;
         }
-
+        
+        /// <summary>
+        /// The client will not throw an exception when the MQTT server responses with a non success ACK packet.
+        /// This will become the default behavior in future versions of the library.
+        /// </summary>
+        public MqttClientOptionsBuilder WithoutThrowOnNonSuccessfulConnectResponse()
+        {
+            _options.ThrowOnNonSuccessfulConnectResponse = false;
+            return this;
+        }
+        
         public MqttClientOptionsBuilder WithAuthentication(string method, byte[] data)
         {
             _options.AuthenticationMethod = method;

Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Net.Security;
 using System.Security.Authentication;
 using System.Security.Cryptography.X509Certificates;
 
@@ -30,11 +31,21 @@ namespace MQTTnet.Client
 #endif
 
 #if NETCOREAPP3_1_OR_GREATER
-        public List<System.Net.Security.SslApplicationProtocol> ApplicationProtocols { get; set; }
-        
-        public System.Net.Security.CipherSuitesPolicy CipherSuitesPolicy { get; set; }
+        public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
+
+        public CipherSuitesPolicy CipherSuitesPolicy { get; set; }
+
+        public EncryptionPolicy EncryptionPolicy { get; set; } = EncryptionPolicy.RequireEncryption;
+
+        public bool AllowRenegotiation { get; set; } = true;
 #endif
 
+        /// <summary>
+        ///     Gets or sets the target host.
+        ///     If the value is null or empty the same host as the TCP socket host will be used.
+        /// </summary>
+        public string TargetHost { get; set; }
+
 #if NET48 || NETCOREAPP3_1_OR_GREATER
         public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12 | SslProtocols.Tls13;
 #else

Source/MQTTnet/Client/Options/MqttClientWebSocketOptions.cs

@@ -2,15 +2,19 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System;
 using System.Collections.Generic;
 using System.Net;
+using System.Net.WebSockets;
 
 namespace MQTTnet.Client
 {
     public sealed class MqttClientWebSocketOptions : IMqttClientChannelOptions
     {
         public CookieContainer CookieContainer { get; set; }
-        
+
+        public ICredentials Credentials { get; set; }
+
         public MqttClientWebSocketProxyOptions ProxyOptions { get; set; }
 
         public IDictionary<string, string> RequestHeaders { get; set; }
@@ -18,12 +22,34 @@ namespace MQTTnet.Client
         public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" };
 
         public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
-        
+
         public string Uri { get; set; }
 
         public override string ToString()
         {
             return Uri;
         }
+
+#if !NETSTANDARD1_3
+#if !WINDOWS_UWP
+        /// <summary>
+        ///     Gets or sets the keep alive interval for the Web Socket connection.
+        ///     This is not related to the keep alive interval for the MQTT protocol.
+        /// </summary>
+        public TimeSpan KeepAliveInterval { get; set; } = WebSocket.DefaultKeepAliveInterval;
+
+        /// <summary>
+        ///     Gets or sets whether the default (system) credentials should be used when connecting via Web Socket connection.
+        ///     This is not related to the credentials which are used for the MQTT protocol.
+        /// </summary>
+        public bool UseDefaultCredentials { get; set; }
+#else
+        /// <summary>
+        ///     Gets or sets the keep alive interval for the Web Socket connection.
+        ///     This is not related to the keep alive interval for the MQTT protocol.
+        /// </summary>
+        public TimeSpan KeepAliveInterval { get; set; } = TimeSpan.FromSeconds(30);
+#endif
+#endif
     }
 }

Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptions.cs

@@ -16,6 +16,8 @@ namespace MQTTnet.Client
 
         public bool BypassOnLocal { get; set; }
 
+        public bool UseDefaultCredentials { get; set; }
+
         public string[] BypassList { get; set; }
     }
 }

Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs

@@ -77,7 +77,7 @@ namespace MQTTnet.Client
 
         internal MqttPublishPacket PublishPacket { get; set; }
 
-        public Task AcknowledgeAsync(CancellationToken cancellationToken)
+        public Task AcknowledgeAsync(CancellationToken cancellationToken = default)
         {
             if (_acknowledgeHandler == null)
             {

Source/MQTTnet/Diagnostics/Runtime/TargetFrameworkProvider.cs

@@ -16,6 +16,8 @@ namespace MQTTnet.Diagnostics
                 return "net461";
 #elif NET472
                 return "net472";
+#elif NET48
+                return "net48";
 #elif NETSTANDARD1_3
                 return "netstandard1.3";
 #elif NETSTANDARD2_0

Source/MQTTnet/Formatter/MqttConnectReasonCodeConverter.cs

@@ -2,54 +2,12 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using MQTTnet.Exceptions;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Formatter
 {
     public static class MqttConnectReasonCodeConverter
     {
-        public static MqttConnectReasonCode ToConnectReasonCode(MqttConnectReturnCode returnCode)
-        {
-            switch (returnCode)
-            {
-                case MqttConnectReturnCode.ConnectionAccepted:
-                {
-                    return MqttConnectReasonCode.Success;
-                }
-
-                case MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion:
-                {
-                    return MqttConnectReasonCode.UnsupportedProtocolVersion;
-                }
-
-                case MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword:
-                {
-                    return MqttConnectReasonCode.BadUserNameOrPassword;
-                }
-
-                case MqttConnectReturnCode.ConnectionRefusedIdentifierRejected:
-                {
-                    return MqttConnectReasonCode.ClientIdentifierNotValid;
-                }
-
-                case MqttConnectReturnCode.ConnectionRefusedServerUnavailable:
-                {
-                    return MqttConnectReasonCode.ServerUnavailable;
-                }
-
-                case MqttConnectReturnCode.ConnectionRefusedNotAuthorized:
-                {
-                    return MqttConnectReasonCode.NotAuthorized;
-                }
-
-                default:
-                {
-                    throw new MqttProtocolViolationException("Unable to convert connect reason code (MQTTv5) to return code (MQTTv3).");
-                }
-            }
-        }
-
         public static MqttConnectReturnCode ToConnectReturnCode(MqttConnectReasonCode reasonCode)
         {
             switch (reasonCode)
@@ -59,11 +17,13 @@ namespace MQTTnet.Formatter
                     return MqttConnectReturnCode.ConnectionAccepted;
                 }
 
+                case MqttConnectReasonCode.Banned:
                 case MqttConnectReasonCode.NotAuthorized:
                 {
                     return MqttConnectReturnCode.ConnectionRefusedNotAuthorized;
                 }
 
+                case MqttConnectReasonCode.BadAuthenticationMethod:
                 case MqttConnectReasonCode.BadUserNameOrPassword:
                 {
                     return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
@@ -79,6 +39,7 @@ namespace MQTTnet.Formatter
                     return MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion;
                 }
 
+                case MqttConnectReasonCode.UseAnotherServer:
                 case MqttConnectReasonCode.ServerUnavailable:
                 case MqttConnectReasonCode.ServerBusy:
                 case MqttConnectReasonCode.ServerMoved:
@@ -87,9 +48,7 @@ namespace MQTTnet.Formatter
                 }
 
                 default:
-                {
-                    throw new MqttProtocolViolationException("Unable to convert connect reason code (MQTTv5) to return code (MQTTv3).");
-                }
+                    return MqttConnectReturnCode.ConnectionRefusedUnacceptableProtocolVersion;
             }
         }
     }

Source/MQTTnet/Formatter/V5/MqttV5PacketDecoder.cs

@@ -130,6 +130,9 @@ namespace MQTTnet.Formatter.V5
                 MaximumQoS = MqttQualityOfServiceLevel.ExactlyOnce
             };
 
+            // Also set the return code of MQTT 3.1.1 for backward compatibility and debugging purposes.
+            packet.ReturnCode = MqttConnectReasonCodeConverter.ToConnectReturnCode(packet.ReasonCode);
+
             var propertiesReader = new MqttV5PropertiesReader(_bufferReader);
             while (propertiesReader.MoveNext())
             {

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -100,6 +100,12 @@ namespace MQTTnet.Implementations
 
                 if (_tcpOptions.TlsOptions?.UseTls == true)
                 {
+                    var targetHost = _tcpOptions.TlsOptions.TargetHost;
+                    if (string.IsNullOrEmpty(targetHost))
+                    {
+                        targetHost = _tcpOptions.Server;
+                    }
+                    
                     var sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
                     try
                     {
@@ -111,14 +117,16 @@ namespace MQTTnet.Implementations
                             EnabledSslProtocols = _tcpOptions.TlsOptions.SslProtocol,
                             CertificateRevocationCheckMode =
  _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode,
-                            TargetHost = _tcpOptions.Server,
-                            CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy
+                            TargetHost = targetHost,
+                            CipherSuitesPolicy = _tcpOptions.TlsOptions.CipherSuitesPolicy,
+                            EncryptionPolicy = _tcpOptions.TlsOptions.EncryptionPolicy,
+                            AllowRenegotiation = _tcpOptions.TlsOptions.AllowRenegotiation
                         };
 
                         await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
 #else
                         await sslStream.AuthenticateAsClientAsync(
-                                _tcpOptions.Server,
+                                targetHost,
                                 LoadCertificates(),
                                 _tcpOptions.TlsOptions.SslProtocol,
                                 !_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors)
@@ -290,12 +298,12 @@ namespace MQTTnet.Implementations
 
         X509CertificateCollection LoadCertificates()
         {
-            var certificates = new X509CertificateCollection();
             if (_tcpOptions.TlsOptions.Certificates == null)
             {
-                return certificates;
+                return null;
             }
 
+            var certificates = new X509CertificateCollection();
             foreach (var certificate in _tcpOptions.TlsOptions.Certificates)
             {
                 certificates.Add(certificate);

Source/MQTTnet/Implementations/MqttTcpServerListener.cs

@@ -231,6 +231,7 @@ namespace MQTTnet.Implementations
                     
                     using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger))
                     {
+                        clientAdapter.AllowPacketFragmentation = _options.AllowPacketFragmentation;
                         await clientHandler(clientAdapter).ConfigureAwait(false);
                     }
                 }
@@ -255,6 +256,7 @@ namespace MQTTnet.Implementations
             {
                 try
                 {
+                    // ReSharper disable once MethodHasAsyncOverload
                     stream?.Dispose();
                     clientSocket?.Dispose();
                 }

Source/MQTTnet/Implementations/MqttWebSocketChannel.cs

@@ -155,14 +155,26 @@ namespace MQTTnet.Implementations
             throw new NotSupportedException("Proxies are not supported when using 'netstandard 1.3'.");
 #else
             var proxyUri = new Uri(_options.ProxyOptions.Address);
-
+            WebProxy webProxy;
+            
             if (!string.IsNullOrEmpty(_options.ProxyOptions.Username) && !string.IsNullOrEmpty(_options.ProxyOptions.Password))
             {
                 var credentials = new NetworkCredential(_options.ProxyOptions.Username, _options.ProxyOptions.Password, _options.ProxyOptions.Domain);
-                return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList, credentials);
+                webProxy = new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList, credentials);
+            }
+            else
+            {
+                webProxy = new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList);    
+            }
+            
+            if (_options.ProxyOptions.UseDefaultCredentials)
+            {
+                // Only update the property if required because setting it to false will alter
+                // the used credentials internally!
+                webProxy.UseDefaultCredentials = true;
             }
 
-            return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList);
+            return webProxy;
 #endif
         }
 
@@ -207,6 +219,26 @@ namespace MQTTnet.Implementations
                 }
             }
 
+#if !NETSTANDARD1_3
+#if !WINDOWS_UWP
+            // Only set the value if it is actually true. This property is not supported on all platforms
+            // and will throw a _PlatformNotSupported_ (i.e. WASM) exception when being used regardless of the actual value.
+            if (_options.UseDefaultCredentials)
+            {
+                clientWebSocket.Options.UseDefaultCredentials = _options.UseDefaultCredentials;
+            }
+
+            if (_options.KeepAliveInterval != WebSocket.DefaultKeepAliveInterval)
+            {
+                clientWebSocket.Options.KeepAliveInterval = _options.KeepAliveInterval;
+            }
+#endif
+#endif
+            if (_options.Credentials != null)
+            {
+                clientWebSocket.Options.Credentials = _options.Credentials;
+            }
+            
             var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler;
             if (certificateValidationHandler != null)
             {
@@ -220,6 +252,8 @@ namespace MQTTnet.Implementations
                 throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'.");
 #elif NET461
                 throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'.");
+#elif NET48
+                throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net48'.");
 #else
                 clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
                 {

Source/MQTTnet/Internal/MqttTaskTimeout.cs

@@ -1,38 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-// See the LICENSE file in the project root for more information.
-
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using MQTTnet.Exceptions;
-
-namespace MQTTnet.Internal
-{
-    public static class MqttTaskTimeout
-    {
-        public static async Task WaitAsync(Func<CancellationToken, Task> action, TimeSpan timeout, CancellationToken cancellationToken)
-        {
-            if (action == null) throw new ArgumentNullException(nameof(action));
-
-            using (var timeoutCts = new CancellationTokenSource(timeout))
-            using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken))
-            {
-                try
-                {
-                    await action(linkedCts.Token).ConfigureAwait(false);
-                }
-                catch (OperationCanceledException exception)
-                {
-                    var timeoutReached = timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested;
-                    if (timeoutReached)
-                    {
-                        throw new MqttCommunicationTimedOutException(exception);
-                    }
-
-                    throw;
-                }
-            }
-        }
-    }
-}

Source/MQTTnet/MQTTnet.csproj

@@ -11,7 +11,7 @@
 
     <PropertyGroup>
         <TargetFrameworks>netstandard1.3;netstandard2.0;netstandard2.1;netcoreapp3.1;net5.0;net6.0;net7.0</TargetFrameworks>
-        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
+        <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461;net48</TargetFrameworks>
         <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core' ">$(TargetFrameworks);uap10.0</TargetFrameworks>
         <LangVersion>7.3</LangVersion>
         <AssemblyName>MQTTnet</AssemblyName>

Source/MQTTnet/MqttApplicationMessageExtensions.cs

@@ -4,6 +4,7 @@
 
 using System;
 using System.Text;
+using MQTTnet.Internal;
 
 namespace MQTTnet
 {
@@ -16,6 +17,16 @@ namespace MQTTnet
                 throw new ArgumentNullException(nameof(applicationMessage));
             }
 
+            if(applicationMessage.PayloadSegment == EmptyBuffer.ArraySegment)
+            {
+                return null;
+            }
+
+            if (applicationMessage.PayloadSegment.Array == null)
+            {
+                return null;
+            }
+
             var payloadSegment = applicationMessage.PayloadSegment;
             return Encoding.UTF8.GetString(payloadSegment.Array, payloadSegment.Offset, payloadSegment.Count);
         }

Source/MQTTnet/Packets/MqttDisconnectPacket.cs

@@ -36,7 +36,7 @@ namespace MQTTnet.Packets
 
         public override string ToString()
         {
-            return $"Disconnect: [ReasonCode={ReasonCode}]";
+            return $"Disconnect: [ReasonCode={ReasonCode}] [ReasonString={ReasonString}] [ServerReference={ServerReference}] [SessionExpiryInterval={SessionExpiryInterval}]";
         }
     }
 }

Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs

@@ -4,26 +4,33 @@
 
 using System;
 using System.Collections;
+using System.Collections.Generic;
 using MQTTnet.Formatter;
+using MQTTnet.Packets;
 
 namespace MQTTnet.Server
 {
     public sealed class ClientConnectedEventArgs : EventArgs
     {
-        public ClientConnectedEventArgs(string clientId, string userName, MqttProtocolVersion protocolVersion, string endpoint, IDictionary sessionItems)
+        readonly MqttConnectPacket _connectPacket;
+
+        public ClientConnectedEventArgs(MqttConnectPacket connectPacket, MqttProtocolVersion protocolVersion, string endpoint, IDictionary sessionItems)
         {
-            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
-            UserName = userName;
+            _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
             ProtocolVersion = protocolVersion;
             Endpoint = endpoint;
             SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
         }
 
+        public byte[] AuthenticationData => _connectPacket.AuthenticationData;
+
+        public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
+
         /// <summary>
         ///     Gets the client identifier of the connected client.
         ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        public string ClientId { get; }
+        public string ClientId => _connectPacket.ClientId;
 
         /// <summary>
         ///     Gets the endpoint of the connected client.
@@ -43,6 +50,12 @@ namespace MQTTnet.Server
         /// <summary>
         ///     Gets the user name of the connected client.
         /// </summary>
-        public string UserName { get; }
+        public string UserName => _connectPacket.Username;
+
+        /// <summary>
+        ///     Gets the user properties sent by the client.
+        ///     <remarks>MQTT 5.0.0+ feature.</remarks>
+        /// </summary>
+        public List<MqttUserProperty> UserProperties => _connectPacket?.UserProperties;
     }
 }

Source/MQTTnet/Server/Internal/MqttClient.cs

@@ -103,6 +103,8 @@ namespace MQTTnet.Server
             finally
             {
                 IsRunning = false;
+                
+                Session.DisconnectedTimestamp = DateTime.UtcNow;
 
                 _cancellationToken?.TryCancel();
                 _cancellationToken?.Dispose();

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -34,7 +34,7 @@ namespace MQTTnet.Server
 
         // The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only.
         // See the MqttSubscription object for a detailed explanation.
-        readonly Dictionary<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(4096);
+        readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage();
 
         readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
         readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>();
@@ -58,7 +58,7 @@ namespace MQTTnet.Server
             _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
         }
 
-        public async Task CloseAllConnectionsAsync()
+        public async Task CloseAllConnections()
         {
             List<MqttClient> connections;
             lock (_clients)
@@ -78,21 +78,16 @@ namespace MQTTnet.Server
             _logger.Verbose("Deleting session for client '{0}'.", clientId);
 
             MqttClient connection;
-
             lock (_clients)
             {
                 _clients.TryGetValue(clientId, out connection);
             }
 
             MqttSession session;
-
             _sessionsManagementLock.EnterWriteLock();
             try
             {
-                _sessions.TryGetValue(clientId, out session);
-                _sessions.Remove(clientId);
-
-                if (session != null)
+                if (_sessionsStorage.TryRemoveSession(clientId, out session))
                 {
                     _subscriberSessions.Remove(session);
                 }
@@ -111,7 +106,7 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, $"Error while deleting session '{clientId}'.");
+                _logger.Error(exception, "Error while deleting session '{0}'", clientId);
             }
 
             try
@@ -124,12 +119,12 @@ namespace MQTTnet.Server
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, $"Error while executing session deleted event for session '{clientId}'.");
+                _logger.Error(exception, "Error while executing session deleted event for session '{0}'", clientId);
             }
 
             session?.Dispose();
 
-            _logger.Verbose("Session for client '{0}' deleted.", clientId);
+            _logger.Verbose("Session of client '{0}' deleted", clientId);
         }
 
         public async Task<DispatchApplicationMessageResult> DispatchApplicationMessage(
@@ -269,20 +264,14 @@ namespace MQTTnet.Server
             _sessionsManagementLock.EnterWriteLock();
             try
             {
-                foreach (var sessionItem in _sessions)
-                {
-                    sessionItem.Value.Dispose();
-                }
+                _sessionsStorage.Dispose();
             }
             finally
             {
                 _sessionsManagementLock.ExitWriteLock();
             }
 
-            if (_sessionsManagementLock != null)
-            {
-                _sessionsManagementLock.Dispose();
-            }
+            _sessionsManagementLock?.Dispose();
         }
 
         public MqttClient GetClient(string id)
@@ -306,17 +295,17 @@ namespace MQTTnet.Server
             }
         }
 
-        public Task<IList<MqttClientStatus>> GetClientStatusesAsync()
+        public Task<IList<MqttClientStatus>> GetClientsStatus()
         {
             var result = new List<MqttClientStatus>();
 
             lock (_clients)
             {
-                foreach (var connection in _clients.Values)
+                foreach (var client in _clients.Values)
                 {
-                    var clientStatus = new MqttClientStatus(connection)
+                    var clientStatus = new MqttClientStatus(client)
                     {
-                        Session = new MqttSessionStatus(connection.Session)
+                        Session = new MqttSessionStatus(client.Session)
                     };
 
                     result.Add(clientStatus);
@@ -326,16 +315,16 @@ namespace MQTTnet.Server
             return Task.FromResult((IList<MqttClientStatus>)result);
         }
 
-        public Task<IList<MqttSessionStatus>> GetSessionStatusAsync()
+        public Task<IList<MqttSessionStatus>> GetSessionsStatus()
         {
             var result = new List<MqttSessionStatus>();
 
             _sessionsManagementLock.EnterReadLock();
             try
             {
-                foreach (var sessionItem in _sessions)
+                foreach (var session in _sessionsStorage.ReadAllSessions())
                 {
-                    var sessionStatus = new MqttSessionStatus(sessionItem.Value);
+                    var sessionStatus = new MqttSessionStatus(session);
                     result.Add(sessionStatus);
                 }
             }
@@ -377,9 +366,7 @@ namespace MQTTnet.Server
 
                 if (_eventContainer.ClientConnectedEvent.HasHandlers)
                 {
-                    var eventArgs = new ClientConnectedEventArgs(
-                        connectPacket.ClientId,
-                        connectPacket.Username,
+                    var eventArgs = new ClientConnectedEventArgs(connectPacket,
                         channelAdapter.PacketFormatterAdapter.ProtocolVersion,
                         channelAdapter.Endpoint,
                         client.Session.Items);
@@ -486,7 +473,7 @@ namespace MQTTnet.Server
         {
             if (!_options.EnablePersistentSessions)
             {
-                _sessions.Clear();
+                _sessionsStorage.Clear();
             }
         }
 
@@ -586,45 +573,49 @@ namespace MQTTnet.Server
                     MqttSession session;
 
                     // Create a new session (if required).
-                    if (!_sessions.TryGetValue(connectPacket.ClientId, out oldSession))
+                    if (!_sessionsStorage.TryGetSession(connectPacket.ClientId, out oldSession))
                     {
-                        session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
+                        session = CreateSession(connectPacket, validatingConnectionEventArgs, sessionShouldPersist);
                     }
                     else
                     {
                         if (connectPacket.CleanSession)
                         {
-                            _logger.Verbose("Deleting existing session of client '{0}' due to clean start.", connectPacket.ClientId);
+                            _logger.Verbose("Deleting existing session of client '{0}' due to clean start", connectPacket.ClientId);
                             _subscriberSessions.Remove(oldSession);
-                            session = CreateSession(connectPacket.ClientId, validatingConnectionEventArgs.SessionItems, sessionShouldPersist);
+                            session = CreateSession(connectPacket, validatingConnectionEventArgs, sessionShouldPersist);
                         }
                         else
                         {
-                            _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
+                            _logger.Verbose("Reusing existing session of client '{0}'", connectPacket.ClientId);
                             session = oldSession;
                             oldSession = null;
 
                             // Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval
                             session.IsPersistent = sessionShouldPersist;
-                            connAckPacket.IsSessionPresent = true;
+                            session.DisconnectedTimestamp = null;
                             session.Recover();
+                            
+                            connAckPacket.IsSessionPresent = true;
                         }
                     }
 
-                    _sessions[connectPacket.ClientId] = session;
-
+                    _sessionsStorage.UpdateSession(connectPacket.ClientId, session);
+                    
                     // Create a new client (always required).
-                    _clients.TryGetValue(connectPacket.ClientId, out oldClient);
-                    if (oldClient != null)
+                    lock (_clients)
                     {
-                        // This will stop the current client from sending and receiving but remains the connection active
-                        // for a later DISCONNECT packet.
-                        oldClient.IsTakenOver = true;
+                        _clients.TryGetValue(connectPacket.ClientId, out oldClient);
+                        if (oldClient != null)
+                        {
+                            // This will stop the current client from sending and receiving but remains the connection active
+                            // for a later DISCONNECT packet.
+                            oldClient.IsTakenOver = true;
+                        }
+
+                        client = CreateClient(connectPacket, channelAdapter, session);
+                        _clients[connectPacket.ClientId] = client;
                     }
-
-                    client = CreateClient(connectPacket, channelAdapter, session);
-
-                    _clients[connectPacket.ClientId] = client;
                 }
                 finally
                 {
@@ -656,11 +647,14 @@ namespace MQTTnet.Server
             return client;
         }
 
-        MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent)
+        MqttSession CreateSession(MqttConnectPacket connectPacket, ValidatingConnectionEventArgs validatingConnectionEventArgs, bool isPersistent)
         {
-            _logger.Verbose("Created new session for client '{0}'.", clientId);
+            _logger.Verbose("Created new session for client '{0}'", connectPacket.ClientId);
 
-            return new MqttSession(clientId, isPersistent, sessionItems, _options, _eventContainer, _retainedMessagesManager, this);
+            return new MqttSession(connectPacket, validatingConnectionEventArgs.SessionItems, _options, _eventContainer, _retainedMessagesManager, this)
+            {
+                IsPersistent = isPersistent
+            };
         }
 
         async Task FireApplicationMessageNotConsumedEvent(MqttApplicationMessage applicationMessage, string senderId)
@@ -679,7 +673,7 @@ namespace MQTTnet.Server
             _sessionsManagementLock.EnterReadLock();
             try
             {
-                if (!_sessions.TryGetValue(clientId, out var session))
+                if (!_sessionsStorage.TryGetSession(clientId, out var session))
                 {
                     throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
                 }

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -162,10 +162,7 @@ namespace MQTTnet.Server
 
         public void Dispose()
         {
-            if (_subscriptionsLock != null)
-            {
-                _subscriptionsLock.Dispose();
-            }
+            _subscriptionsLock?.Dispose();
         }
 
         public async Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)

Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs

@@ -134,9 +134,9 @@ namespace MQTTnet.Server
                 {
                     return Task.FromResult(message);
                 }
-
-                return null;
             }
+
+            return Task.FromResult<MqttApplicationMessage>(null);
         }
 
         public async Task ClearMessages()

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -21,6 +21,7 @@ namespace MQTTnet.Server
         readonly MqttPacketBus _packetBus = new MqttPacketBus();
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
+        readonly MqttConnectPacket _connectPacket;
         readonly MqttServerOptions _serverOptions;
         readonly MqttClientSubscriptionsManager _subscriptionsManager;
 
@@ -31,18 +32,16 @@ namespace MQTTnet.Server
         HashSet<string> _subscribedTopics;
 
         public MqttSession(
-            string clientId,
-            bool isPersistent,
+            MqttConnectPacket connectPacket,
             IDictionary items,
             MqttServerOptions serverOptions,
             MqttServerEventContainer eventContainer,
             MqttRetainedMessagesManager retainedMessagesManager,
             MqttClientSessionsManager clientSessionsManager)
         {
-            Id = clientId ?? throw new ArgumentNullException(nameof(clientId));
-            IsPersistent = isPersistent;
             Items = items ?? throw new ArgumentNullException(nameof(items));
 
+            _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
 
@@ -50,10 +49,14 @@ namespace MQTTnet.Server
         }
 
         public DateTime CreatedTimestamp { get; } = DateTime.UtcNow;
+        
+        public DateTime? DisconnectedTimestamp { get; set; }
 
         public bool HasSubscribedTopics => _subscribedTopics != null && _subscribedTopics.Count > 0;
 
-        public string Id { get; }
+        public uint ExpiryInterval => _connectPacket.SessionExpiryInterval;
+        
+        public string Id => _connectPacket.ClientId;
 
         /// <summary>
         ///     Session should persist if CleanSession was set to false (Mqtt3) or if SessionExpiryInterval != 0 (Mqtt5)

Source/MQTTnet/Server/Internal/MqttSessionsStorage.cs

@@ -0,0 +1,93 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Collections.Generic;
+using System.Collections.ObjectModel;
+using System.Linq;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttSessionsStorage
+    {
+        readonly Dictionary<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(4096);
+
+        public void Clear()
+        {
+            // Make sure that the sessions are also getting disposed!
+            // This object can be reused after disposing.
+            Dispose();
+        }
+
+        public void Dispose()
+        {
+            foreach (var sessionItem in _sessions)
+            {
+                sessionItem.Value.Dispose();
+            }
+
+            _sessions.Clear();
+        }
+
+        public IReadOnlyCollection<MqttSession> ReadAllSessions()
+        {
+            return new ReadOnlyCollection<MqttSession>(_sessions.Values.ToList());
+        }
+
+        public bool TryGetSession(string id, out MqttSession session)
+        {
+            if (id == null)
+            {
+                throw new ArgumentNullException(nameof(id));
+            }
+
+            if (!_sessions.TryGetValue(id, out session))
+            {
+                return false;
+            }
+
+            // If the Session Expiry Interval is 0xFFFFFFFF (UINT_MAX), the Session does not expire.
+            if (session.ExpiryInterval > 0 && session.ExpiryInterval != uint.MaxValue)
+            {
+                if (session.DisconnectedTimestamp.HasValue)
+                {
+                    if (DateTime.UtcNow > session.DisconnectedTimestamp.Value.AddSeconds(session.ExpiryInterval))
+                    {
+                        TryRemoveSession(id, out var oldSession);
+                        oldSession?.Dispose();
+                        return false;
+                    }
+                }
+            }
+
+            return true;
+        }
+
+        public bool TryRemoveSession(string id, out MqttSession session)
+        {
+            if (id == null)
+            {
+                throw new ArgumentNullException(nameof(id));
+            }
+
+            if (_sessions.TryGetValue(id, out session))
+            {
+                _sessions.Remove(id);
+                return true;
+            }
+
+            return false;
+        }
+
+        public void UpdateSession(string id, MqttSession session)
+        {
+            if (id == null)
+            {
+                throw new ArgumentNullException(nameof(id));
+            }
+
+            _sessions[id] = session;
+        }
+    }
+}

Source/MQTTnet/Server/MqttServer.cs

@@ -200,7 +200,7 @@ namespace MQTTnet.Server
         {
             ThrowIfNotStarted();
 
-            return _clientSessionsManager.GetClientStatusesAsync();
+            return _clientSessionsManager.GetClientsStatus();
         }
 
         public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
@@ -226,10 +226,16 @@ namespace MQTTnet.Server
         {
             ThrowIfNotStarted();
 
-            return _clientSessionsManager.GetSessionStatusAsync();
+            return _clientSessionsManager.GetSessionsStatus();
         }
 
+        [Obsolete("Use InjectApplicationMessageAsync instead.")]
         public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
+        {
+            return InjectApplicationMessageAsync(injectedApplicationMessage, cancellationToken);
+        }
+        
+        public Task InjectApplicationMessageAsync(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
         {
             if (injectedApplicationMessage == null)
             {
@@ -292,7 +298,7 @@ namespace MQTTnet.Server
 
                 _cancellationTokenSource.Cancel(false);
 
-                await _clientSessionsManager.CloseAllConnectionsAsync().ConfigureAwait(false);
+                await _clientSessionsManager.CloseAllConnections().ConfigureAwait(false);
 
                 foreach (var adapter in _adapters)
                 {

Source/MQTTnet/Server/MqttServerExtensions.cs

@@ -13,12 +13,38 @@ namespace MQTTnet.Server
 {
     public static class MqttServerExtensions
     {
+        [Obsolete("Use InjectApplicationMessageAsync instead.")]
         public static Task InjectApplicationMessage(
             this MqttServer server,
             string topic,
             string payload = null,
             MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
             bool retain = false)
+        {
+            return InjectApplicationMessageAsync(server, topic, payload, qualityOfServiceLevel, retain);
+        }
+
+        public static Task InjectApplicationMessageAsync(this MqttServer server, MqttApplicationMessage applicationMessage)
+        {
+            if (server == null)
+            {
+                throw new ArgumentNullException(nameof(server));
+            }
+
+            if (applicationMessage == null)
+            {
+                throw new ArgumentNullException(nameof(applicationMessage));
+            }
+
+            return server.InjectApplicationMessageAsync(new InjectedMqttApplicationMessage(applicationMessage));
+        }
+
+        public static Task InjectApplicationMessageAsync(
+            this MqttServer server,
+            string topic,
+            string payload = null,
+            MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce,
+            bool retain = false)
         {
             if (server == null)
             {
@@ -36,7 +62,7 @@ namespace MQTTnet.Server
                 payloadBuffer = Encoding.UTF8.GetBytes(stringPayload);
             }
 
-            return server.InjectApplicationMessage(
+            return server.InjectApplicationMessageAsync(
                 new InjectedMqttApplicationMessage(
                     new MqttApplicationMessage
                     {

Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs

@@ -131,6 +131,18 @@ namespace MQTTnet.Server
             return this;
         }
 
+        /// <summary>
+        ///     Usually the MQTT packets can be send partially to improve memory allocations.
+        ///     This is done by using multiple TCP packets or WebSocket frames etc.
+        ///     Unfortunately not all clients do support this and will close the connection when receiving partial packets.
+        /// </summary>
+        public MqttServerOptionsBuilder WithoutPacketFragmentation()
+        {
+            _options.DefaultEndpointOptions.AllowPacketFragmentation = false;
+            _options.TlsEndpointOptions.AllowPacketFragmentation = false;
+            return this;
+        }
+
         public MqttServerOptionsBuilder WithPersistentSessions(bool value = true)
         {
             _options.EnablePersistentSessions = value;

Source/MQTTnet/Server/Options/MqttServerTcpEndpointBaseOptions.cs

@@ -23,6 +23,14 @@ namespace MQTTnet.Server
         /// </summary>
         public bool? KeepAlive { get; set; }
 
+        /// <summary>
+        ///     Usually the MQTT packets can be send partially. This is done by using multiple TCP packets
+        ///     or WebSocket frames etc. Unfortunately not all clients do support this feature and
+        ///     will close the connection when receiving such packets. If such clients are connecting to this
+        ///     server the flag must be set to _false_.
+        /// </summary>
+        public bool AllowPacketFragmentation { get; set; } = true;
+
         /// <summary>
         ///     Gets or sets the TCP keep alive interval.
         ///     The value _null_ indicates that the OS and framework defaults should be used.

Source/MQTTnet/Server/Status/MqttSessionStatus.cs

@@ -21,6 +21,10 @@ namespace MQTTnet.Server
 
         public DateTime CreatedTimestamp => _session.CreatedTimestamp;
 
+        public DateTime? DisconnectedTimestamp => _session.DisconnectedTimestamp;
+
+        public uint ExpiryInterval => _session.ExpiryInterval;
+
         public string Id => _session.Id;
 
         public IDictionary Items => _session.Items;