Add samples.

commit 469caf656e
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Sun Jul 31 10:55:40 2022 +0200

    Fix build errors.

commit 42402fba04
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Sun Jul 31 10:22:23 2022 +0200

    Allow interception of injected application messages.

commit a0dc8ab0d7
Author: William Lee <31092948+william19821101@users.noreply.github.com>
Date:   Sun Jul 31 15:50:14 2022 +0800

    Update Server_Simple_Samples.cs (#1474)

    Change e.UserName to e.Username

.github/workflows/ReleaseNotes.md

@@ -1,8 +0,0 @@
-* [Core] Updated nuget packages.
-* [Core] The option _IgnoreCertificateChainErrors_ is now respected (thanks to @GodVenn, #1447).
-* [ManagedClient] The managed client now sends the entire topic filter including new MQTTv5 properties when subscribing.
-* [ManagedClient] Fixed wrong firing of _ApplicationMessageSkippedAsync_ event (thanks to @quackgizmo, #1460).
-* [Server] Fixed reporting of _MaximumQoS_ in _ConnAck_ packet (MQTTv5 only) (#1442).
-* [Server] Fix cross thread issue in session message storage for QoS 1 and 2.
-* [Server] The event _ClientSubscribedTopicAsync_ is now fired after the subscription is completely processed internally (#1435).
-* [Server] Improved CPU usage on lower end machines (#788).

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "4.0.2.${{github.run_number}}"
+  VERSION: "4.1.0.${{github.run_number}}"
 
 jobs:
   build:

MQTTnet.sln.DotSettings

@@ -29,7 +29,7 @@
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/XmlDocFormatter/ProcessingInstructionAttributesFormat/@EntryValue">OnDifferentLines</s:String>
 	<s:String x:Key="/Default/CodeStyle/CodeFormatting/XmlDocFormatter/TagAttributesFormat/@EntryValue">OnDifferentLines</s:String>
 	<s:String x:Key="/Default/CodeStyle/CSharpFileLayoutPatterns/Pattern/@EntryValue">&lt;?xml version="1.0" encoding="utf-16"?&gt;
-&lt;Patterns xmlns="urn:schemas-jetbrains-com:member-reordering-patterns"&gt;
+&lt;Patterns xmlns="urn:schemas-jetbrains-com:member-reordering-patterns"&gt;    
   &lt;TypePattern DisplayName="Non-reorderable types"&gt;
     &lt;TypePattern.Match&gt;
       &lt;Or&gt;
@@ -226,6 +226,9 @@
     &lt;/Entry&gt;
   &lt;/TypePattern&gt;
 &lt;/Patterns&gt;</s:String>
+	<s:String x:Key="/Default/CodeStyle/FileHeader/FileHeaderText/@EntryValue">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.</s:String>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>

Samples/Client/Client_Connection_Samples.cs

@@ -2,10 +2,15 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using System.Security.Authentication;
 using MQTTnet.Client;
+using MQTTnet.Extensions.WebSocket4Net;
 using MQTTnet.Formatter;
-using MQTTnet.Samples.Helpers;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Client;
 
@@ -150,6 +155,30 @@ public static class Client_Connection_Samples
             response.DumpToConsole();
         }
     }
+    
+    public static async Task Connect_Client_Using_WebSocket4Net()
+    {
+        /*
+         * This sample creates a simple MQTT client and connects to a public broker using a WebSocket connection.
+         * Instead of the .NET implementation of WebSockets the implementaion from WebSocket4Net is used. It provides more
+         * encryption algorithms and supports more platforms.
+         * 
+         * This is a modified version of the sample _Connect_Client_! See other sample for more details.
+         */
+
+        var mqttFactory = new MqttFactory().UseWebSocket4Net();
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            var mqttClientOptions = new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").Build();
+
+            var response = await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+            Console.WriteLine("The MQTT client is connected.");
+
+            response.DumpToConsole();
+        }
+    }
 
     public static async Task Connect_Client_With_TLS_Encryption()
     {

Samples/Client/Client_Publish_Samples.cs

@@ -2,6 +2,10 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using MQTTnet.Client;
 
 namespace MQTTnet.Samples.Client;

Samples/Client/Client_Subscribe_Samples.cs

@@ -2,8 +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.
 
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using MQTTnet.Client;
-using MQTTnet.Samples.Helpers;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Client;
 

Samples/Diagnostics/Logger_Samples.cs

@@ -2,8 +2,13 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using System.Text;
 using MQTTnet.Diagnostics;
+using MQTTnet.Samples.Shared;
 
 namespace MQTTnet.Samples.Diagnostics;
 
@@ -18,9 +23,7 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(new MyLogger());
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
-            .WithTcpServer("broker.hivemq.com")
-            .Build();
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -28,13 +31,31 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
-                .Build();
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
 
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
     }
 
+    public static async Task Log_To_Console()
+    {
+        /*
+       * This sample covers the creation of a custom logger which is used to log to the console.
+       */
+
+        var mqttFactory = new MqttFactory(new ConsoleLogger());
+
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+
+        using (var mqttClient = mqttFactory.CreateMqttClient())
+        {
+            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+            Console.WriteLine("MQTT client is connected. Press any key to exit.");
+            Console.ReadKey(true);
+        }
+    }
+
     public static async Task Use_Event_Logger()
     {
         /*
@@ -60,9 +81,7 @@ public static class Logger_Samples
 
         var mqttFactory = new MqttFactory(mqttEventLogger);
 
-        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder()
-            .WithTcpServer("broker.hivemq.com")
-            .Build();
+        var mqttClientOptions = mqttFactory.CreateClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
 
         using (var mqttClient = mqttFactory.CreateMqttClient())
         {
@@ -70,8 +89,7 @@ public static class Logger_Samples
 
             Console.WriteLine("MQTT client is connected.");
 
-            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder()
-                .Build();
+            var mqttClientDisconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
 
             await mqttClient.DisconnectAsync(mqttClientDisconnectOptions, CancellationToken.None);
         }
@@ -79,7 +97,7 @@ public static class Logger_Samples
 
     sealed class MyLogger : IMqttNetLogger
     {
-        public bool IsEnabled { get; set; } = true;
+        public bool IsEnabled { get; } = true;
 
         public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
         {

Samples/Diagnostics/PackageInspection_Samples.cs

@@ -2,6 +2,10 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using MQTTnet.Diagnostics;
 
 namespace MQTTnet.Samples.Diagnostics;

Samples/ManagedClient/Managed_Client_Simple_Samples.cs

@@ -1,3 +1,11 @@
+// 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.
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using MQTTnet.Client;
 using MQTTnet.Extensions.ManagedClient;
 

Samples/Program.cs

@@ -8,6 +8,7 @@ Console.WriteLine("Welcome to MQTTnet samples!");
 Console.WriteLine();
 
 var sampleClasses = Assembly.GetExecutingAssembly().GetExportedTypes().OrderBy(c => c.Name).ToList();
+sampleClasses.RemoveAll(c => c.Namespace.Contains(".Shared"));
 
 var index = 0;
 foreach (var sampleClass in sampleClasses)

Samples/RpcClient/RpcClient_Samples.cs

@@ -2,6 +2,10 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using MQTTnet.Client;
 using MQTTnet.Extensions.Rpc;
 using MQTTnet.Protocol;

Samples/Server/Server_Asp_Net_Samples.cs

@@ -0,0 +1,110 @@
+// 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.
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+// ReSharper disable EmptyConstructor
+// ReSharper disable MemberCanBeMadeStatic.Local
+
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using MQTTnet.AspNetCore;
+using MQTTnet.Server;
+
+namespace MQTTnet.Samples.Server;
+
+public static class Server_Asp_Net_Samples
+{
+    public static Task Start_Server_With_WebSockets_Support()
+    {
+        /*
+         * This sample starts a minimal ASP.NET Webserver including a hosted MQTT server.
+         */
+        var host = Host.CreateDefaultBuilder(Array.Empty<string>())
+            .ConfigureWebHostDefaults(
+                webBuilder =>
+                {
+                    webBuilder.UseKestrel(
+                        o =>
+                        {
+                            // This will allow MQTT connections based on TCP port 1883.
+                            o.ListenAnyIP(1883, l => l.UseMqtt());
+
+                            // This will allow MQTT connections based on HTTP WebSockets with URI "localhost:5000/mqtt"
+                            // See code below for URI configuration.
+                            o.ListenAnyIP(5000); // Default HTTP pipeline
+                        });
+
+                    webBuilder.UseStartup<Startup>();
+                });
+
+        return host.RunConsoleAsync();
+    }
+
+    sealed class MqttController
+    {
+        public MqttController()
+        {
+            // Inject other services via constructor.
+        }
+        
+        public Task OnClientConnected(ClientConnectedEventArgs eventArgs)
+        {
+            Console.WriteLine($"Client '{eventArgs.ClientId}' connected.");
+            return Task.CompletedTask;
+        }
+
+
+        public Task ValidateConnection(ValidatingConnectionEventArgs eventArgs)
+        {
+            Console.WriteLine($"Client '{eventArgs.ClientId}' wants to connect. Accepting!");
+            return Task.CompletedTask;
+        }
+    }
+
+    sealed class Startup
+    {
+        public void Configure(IApplicationBuilder app, IWebHostEnvironment environment, MqttController mqttController)
+        {
+            app.UseRouting();
+
+            app.UseEndpoints(
+                endpoints =>
+                {
+                    endpoints.MapConnectionHandler<MqttConnectionHandler>(
+                        "/mqtt",
+                        httpConnectionDispatcherOptions => httpConnectionDispatcherOptions.WebSockets.SubProtocolSelector =
+                            protocolList => protocolList.FirstOrDefault() ?? string.Empty);
+                });
+
+            app.UseMqttServer(
+                server =>
+                {
+                    /*
+                     * Attach event handlers etc. if required.
+                     */
+
+                    server.ValidatingConnectionAsync += mqttController.ValidateConnection;
+                    server.ClientConnectedAsync += mqttController.OnClientConnected;
+                });
+        }
+
+        public void ConfigureServices(IServiceCollection services)
+        {
+            services.AddHostedMqttServer(
+                optionsBuilder =>
+                {
+                    optionsBuilder.WithDefaultEndpoint();
+                });
+
+            services.AddMqttConnectionHandler();
+            services.AddConnections();
+
+            services.AddSingleton<MqttController>();
+        }
+    }
+}

Samples/Server/Server_Retained_Messages_Samples.cs

@@ -1,3 +1,11 @@
+// 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.
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using System.Text.Json;
 
 namespace MQTTnet.Samples.Server;

Samples/Server/Server_Simple_Samples.cs

@@ -2,8 +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.Diagnostics;
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
 using MQTTnet.Protocol;
+using MQTTnet.Samples.Shared;
 using MQTTnet.Server;
 
 namespace MQTTnet.Samples.Server;
@@ -134,7 +138,7 @@ public static class Server_Simple_Samples
                     e.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
                 }
 
-                if (e.Username != "ValidUser")
+                if (e.UserName != "ValidUser")
                 {
                     e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                 }
@@ -166,50 +170,4 @@ public static class Server_Simple_Samples
         await server.StartAsync();
         return server;
     }
-
-    class ConsoleLogger : IMqttNetLogger
-    {
-        readonly object _consoleSyncRoot = new();
-
-        public bool IsEnabled => true;
-
-        public void Publish(MqttNetLogLevel logLevel, string source, string message, object[]? parameters, Exception? exception)
-        {
-            var foregroundColor = ConsoleColor.White;
-            switch (logLevel)
-            {
-                case MqttNetLogLevel.Verbose:
-                    foregroundColor = ConsoleColor.White;
-                    break;
-
-                case MqttNetLogLevel.Info:
-                    foregroundColor = ConsoleColor.Green;
-                    break;
-
-                case MqttNetLogLevel.Warning:
-                    foregroundColor = ConsoleColor.DarkYellow;
-                    break;
-
-                case MqttNetLogLevel.Error:
-                    foregroundColor = ConsoleColor.Red;
-                    break;
-            }
-
-            if (parameters?.Length > 0)
-            {
-                message = string.Format(message, parameters);
-            }
-
-            lock (_consoleSyncRoot)
-            {
-                Console.ForegroundColor = foregroundColor;
-                Console.WriteLine(message);
-
-                if (exception != null)
-                {
-                    Console.WriteLine(exception);
-                }
-            }
-        }
-    }
-}
+}

Samples/Shared/ConsoleLogger.cs

@@ -0,0 +1,53 @@
+// 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 MQTTnet.Diagnostics;
+
+namespace MQTTnet.Samples.Shared;
+
+public sealed class ConsoleLogger : IMqttNetLogger
+{
+    readonly object _consoleSyncRoot = new();
+
+    public bool IsEnabled => true;
+
+    public void Publish(MqttNetLogLevel logLevel, string source, string message, object[]? parameters, Exception? exception)
+    {
+        var foregroundColor = ConsoleColor.White;
+        switch (logLevel)
+        {
+            case MqttNetLogLevel.Verbose:
+                foregroundColor = ConsoleColor.White;
+                break;
+
+            case MqttNetLogLevel.Info:
+                foregroundColor = ConsoleColor.Green;
+                break;
+
+            case MqttNetLogLevel.Warning:
+                foregroundColor = ConsoleColor.DarkYellow;
+                break;
+
+            case MqttNetLogLevel.Error:
+                foregroundColor = ConsoleColor.Red;
+                break;
+        }
+
+        if (parameters?.Length > 0)
+        {
+            message = string.Format(message, parameters);
+        }
+
+        lock (_consoleSyncRoot)
+        {
+            Console.ForegroundColor = foregroundColor;
+            Console.WriteLine(message);
+
+            if (exception != null)
+            {
+                Console.WriteLine(exception);
+            }
+        }
+    }
+}

Samples/Shared/ObjectExtensions.cs

@@ -4,7 +4,7 @@
 
 using System.Text.Json;
 
-namespace MQTTnet.Samples.Helpers;
+namespace MQTTnet.Samples.Shared;
 
 internal static class ObjectExtensions
 {

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs

@@ -98,7 +98,6 @@ namespace MQTTnet.Extensions.WebSocket4Net
 #else
                     certificates.Add(certificate);
 #endif
-                    
                 }
             }
 

Source/MQTTnet.TestApp/ServerTest.cs

@@ -118,7 +118,7 @@ namespace MQTTnet.TestApp
                 {
                     if (e.ClientId == "SpecialClient")
                     {
-                        if (e.Username != "USER" || e.Password != "PASS")
+                        if (e.UserName != "USER" || e.Password != "PASS")
                         {
                             e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                         }

Source/MQTTnet.Tests/Internal/AsyncEvent_Tests.cs

@@ -0,0 +1,66 @@
+// 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.Implementations;
+using MQTTnet.Internal;
+
+namespace MQTTnet.Tests.Internal
+{
+    [TestClass]
+    public sealed class AsyncEvent_Tests
+    {
+        int _testEventAsyncCount;
+
+        [TestMethod]
+        public async Task Attach_EventHandler()
+        {
+            var testClass = new TestClass();
+
+            testClass.TestEventAsync += OnTestEventAsync;
+
+            await testClass.FireEventAsync(EventArgs.Empty);
+
+            Assert.AreEqual(1, _testEventAsyncCount);
+        }
+
+        [TestMethod]
+        public async Task Detach_EventHandler()
+        {
+            var testClass = new TestClass();
+
+            testClass.TestEventAsync += OnTestEventAsync;
+            testClass.TestEventAsync -= OnTestEventAsync;
+
+            await testClass.FireEventAsync(EventArgs.Empty);
+
+            Assert.AreEqual(0, _testEventAsyncCount);
+        }
+
+        Task OnTestEventAsync(EventArgs arg)
+        {
+            Interlocked.Increment(ref _testEventAsyncCount);
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
+        sealed class TestClass
+        {
+            readonly AsyncEvent<EventArgs> _testEvent = new AsyncEvent<EventArgs>();
+
+            public event Func<EventArgs, Task> TestEventAsync
+            {
+                add => _testEvent.AddHandler(value);
+                remove => _testEvent.RemoveHandler(value);
+            }
+
+            public Task FireEventAsync(EventArgs eventArgs)
+            {
+                return _testEvent.InvokeAsync(eventArgs);
+            }
+        }
+    }
+}

Source/MQTTnet.Tests/Server/General.cs

@@ -798,38 +798,7 @@ namespace MQTTnet.Tests.Server
                 Assert.IsTrue(longBody.SequenceEqual(receivedBody ?? new byte[0]));
             }
         }
-
-        [TestMethod]
-        public async Task Session_Takeover()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                await testEnvironment.StartServer();
-
-                var options = new MqttClientOptionsBuilder().WithCleanSession(false)
-                    .WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+
-                    .WithClientId("a");
-
-                var client1 = await testEnvironment.ConnectClient(options);
-                await Task.Delay(500);
-
-                var disconnectReason = MqttClientDisconnectReason.NormalDisconnection;
-                client1.DisconnectedAsync += c =>
-                {
-                    disconnectReason = c.Reason;
-                    return PlatformAbstractionLayer.CompletedTask;
-                };
-
-                var client2 = await testEnvironment.ConnectClient(options);
-                await Task.Delay(500);
-
-                Assert.IsFalse(client1.IsConnected);
-                Assert.IsTrue(client2.IsConnected);
-
-                Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason);
-            }
-        }
-
+        
         [TestMethod]
         public async Task Set_Subscription_At_Server()
         {
@@ -944,26 +913,6 @@ namespace MQTTnet.Tests.Server
             }
         }
 
-        [TestMethod]
-        public async Task Use_Clean_Session()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                await testEnvironment.StartServer();
-
-                var client = testEnvironment.CreateClient();
-                var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build());
-
-                // Create the session including the subscription.
-                var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false));
-                await client1.SubscribeAsync("x");
-                await client1.DisconnectAsync();
-                await Task.Delay(500);
-
-                Assert.IsFalse(connectResult.IsSessionPresent);
-            }
-        }
-
         [TestMethod]
         public async Task Use_Empty_Client_ID()
         {
@@ -986,98 +935,7 @@ namespace MQTTnet.Tests.Server
                 Assert.IsTrue(client.IsConnected);
             }
         }
-
-        [TestMethod]
-        public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                var receivedMessagesCount = 0;
-
-                await testEnvironment.StartServer();
-
-                var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will");
-
-                var c1 = await testEnvironment.ConnectClient();
-
-                c1.ApplicationMessageReceivedAsync += e =>
-                {
-                    Interlocked.Increment(ref receivedMessagesCount);
-                    return PlatformAbstractionLayer.CompletedTask;
-                };
-
-                await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
-
-                var c2 = await testEnvironment.ConnectClient(clientOptions);
-                await c2.DisconnectAsync().ConfigureAwait(false);
-
-                await Task.Delay(1000);
-
-                Assert.AreEqual(0, receivedMessagesCount);
-            }
-        }
-
-        [TestMethod]
-        public async Task Will_Message_Do_Not_Send_On_Takeover()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                var receivedMessagesCount = 0;
-
-                await testEnvironment.StartServer();
-
-                // C1 will receive the last will!
-                var c1 = await testEnvironment.ConnectClient();
-                c1.ApplicationMessageReceivedAsync += e =>
-                {
-                    Interlocked.Increment(ref receivedMessagesCount);
-                    return PlatformAbstractionLayer.CompletedTask;
-                };
-
-                await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
-
-                // C2 has the last will defined.
-                var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithClientId("WillOwner");
-
-                var c2 = await testEnvironment.ConnectClient(clientOptions);
-
-                // C3 will do the connection takeover.
-                var c3 = await testEnvironment.ConnectClient(clientOptions);
-
-                await Task.Delay(1000);
-
-                Assert.AreEqual(0, receivedMessagesCount);
-            }
-        }
-
-        [TestMethod]
-        public async Task Will_Message_Send()
-        {
-            using (var testEnvironment = CreateTestEnvironment())
-            {
-                await testEnvironment.StartServer();
-
-                var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder());
-
-                var receivedMessagesCount = 0;
-                c1.ApplicationMessageReceivedAsync += e =>
-                {
-                    Interlocked.Increment(ref receivedMessagesCount);
-                    return PlatformAbstractionLayer.CompletedTask;
-                };
-
-                await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
-
-                var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce);
-                var c2 = await testEnvironment.ConnectClient(clientOptions);
-                c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
-
-                await Task.Delay(1000);
-
-                Assert.AreEqual(1, receivedMessagesCount);
-            }
-        }
-
+        
         void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs)
         {
             if (_connected.ContainsKey(eventArgs.ClientId))

Source/MQTTnet.Tests/Server/QoS_Tests.cs

@@ -0,0 +1,124 @@
+// 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.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Implementations;
+using MQTTnet.Protocol;
+using MQTTnet.Server;
+
+namespace MQTTnet.Tests.Server
+{
+    [TestClass]
+    public sealed class QoS_Tests : BaseTestClass
+    {
+        [TestMethod]
+        public async Task Fire_Event_On_Client_Acknowledges_QoS_0()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                ClientAcknowledgedPublishPacketEventArgs eventArgs = null;
+                server.ClientAcknowledgedPublishPacketAsync += args =>
+                {
+                    eventArgs = args;
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                var client1 = await testEnvironment.ConnectClient();
+                await client1.SubscribeAsync("A");
+
+                var client2 = await testEnvironment.ConnectClient();
+                await client2.PublishStringAsync("A");
+
+                await LongTestDelay();
+
+                // Must be null because no event should be fired for QoS 0.
+                Assert.IsNull(eventArgs);
+            }
+        }
+
+        [TestMethod]
+        public async Task Fire_Event_On_Client_Acknowledges_QoS_1()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                ClientAcknowledgedPublishPacketEventArgs eventArgs = null;
+                server.ClientAcknowledgedPublishPacketAsync += args =>
+                {
+                    eventArgs = args;
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                var client1 = await testEnvironment.ConnectClient();
+                await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.AtLeastOnce);
+
+                var client2 = await testEnvironment.ConnectClient();
+                await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.AtLeastOnce);
+
+                await LongTestDelay();
+
+                Assert.IsNotNull(eventArgs);
+                Assert.IsNotNull(eventArgs.PublishPacket);
+                Assert.IsNotNull(eventArgs.AcknowledgePacket);
+                Assert.IsTrue(eventArgs.IsCompleted);
+
+                Assert.AreEqual("A", eventArgs.PublishPacket.Topic);
+            }
+        }
+
+        [TestMethod]
+        public async Task Fire_Event_On_Client_Acknowledges_QoS_2()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var server = await testEnvironment.StartServer();
+
+                var eventArgs = new List<ClientAcknowledgedPublishPacketEventArgs>();
+                server.ClientAcknowledgedPublishPacketAsync += args =>
+                {
+                    lock (eventArgs)
+                    {
+                        eventArgs.Add(args);
+                    }
+
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                var client1 = await testEnvironment.ConnectClient();
+                await client1.SubscribeAsync("A", MqttQualityOfServiceLevel.ExactlyOnce);
+
+                var client2 = await testEnvironment.ConnectClient();
+                await client2.PublishStringAsync("A", qualityOfServiceLevel: MqttQualityOfServiceLevel.ExactlyOnce);
+
+                await LongTestDelay();
+
+                Assert.AreEqual(2, eventArgs.Count);
+
+                var firstEvent = eventArgs[0];
+
+                Assert.IsNotNull(firstEvent);
+                Assert.IsNotNull(firstEvent.PublishPacket);
+                Assert.IsNotNull(firstEvent.AcknowledgePacket);
+                Assert.IsFalse(firstEvent.IsCompleted);
+
+                Assert.AreEqual("A", firstEvent.PublishPacket.Topic);
+
+                var secondEvent = eventArgs[1];
+
+                Assert.IsNotNull(secondEvent);
+                Assert.IsNotNull(secondEvent.PublishPacket);
+                Assert.IsNotNull(secondEvent.AcknowledgePacket);
+                Assert.IsTrue(secondEvent.IsCompleted);
+
+                Assert.AreEqual("A", secondEvent.PublishPacket.Topic);
+            }
+        }
+    }
+}

Source/MQTTnet.Tests/Server/Security_Tests.cs

@@ -2,10 +2,8 @@
 // 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.Threading.Tasks;
-using Microsoft.CodeAnalysis;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Adapter;
 using MQTTnet.Client;
@@ -40,7 +38,7 @@ namespace MQTTnet.Tests.Server
 
                 testEnvironment.Server.ValidatingConnectionAsync += eventArgs =>
                 {
-                    if (eventArgs.Username == "SECRET")
+                    if (eventArgs.UserName == "SECRET")
                     {
                         eventArgs.ReasonCode = MqttConnectReasonCode.Success;
                     }
@@ -149,7 +147,7 @@ namespace MQTTnet.Tests.Server
 
                 server.ValidatingConnectionAsync += e =>
                 {
-                    if (e.Username != "UserName1")
+                    if (e.UserName != "UserName1")
                     {
                         e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                     }

Source/MQTTnet.Tests/Server/Session_Tests.cs

@@ -14,7 +14,6 @@ using MQTTnet.Implementations;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
-using MqttClient = MQTTnet.Client.MqttClient;
 
 namespace MQTTnet.Tests.Server
 {
@@ -220,6 +219,37 @@ namespace MQTTnet.Tests.Server
             }
         }
 
+        [TestMethod]
+        public async Task Session_Takeover()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var options = new MqttClientOptionsBuilder().WithCleanSession(false)
+                    .WithProtocolVersion(MqttProtocolVersion.V500) // Disconnect reason is only available in MQTT 5+
+                    .WithClientId("a");
+
+                var client1 = await testEnvironment.ConnectClient(options);
+                await Task.Delay(500);
+
+                var disconnectReason = MqttClientDisconnectReason.NormalDisconnection;
+                client1.DisconnectedAsync += c =>
+                {
+                    disconnectReason = c.Reason;
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                var client2 = await testEnvironment.ConnectClient(options);
+                await Task.Delay(500);
+
+                Assert.IsFalse(client1.IsConnected);
+                Assert.IsTrue(client2.IsConnected);
+
+                Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason);
+            }
+        }
+
         [TestMethod]
         public async Task Set_Session_Item()
         {
@@ -277,6 +307,59 @@ namespace MQTTnet.Tests.Server
             }
         }
 
+        [TestMethod]
+        public async Task Use_Clean_Session()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var client = testEnvironment.CreateClient();
+                var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build());
+
+                // Create the session including the subscription.
+                var client1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("a").WithCleanSession(false));
+                await client1.SubscribeAsync("x");
+                await client1.DisconnectAsync();
+                await Task.Delay(500);
+
+                Assert.IsFalse(connectResult.IsSessionPresent);
+            }
+        }
+
+        [TestMethod]
+        public async Task Will_Message_Do_Not_Send_On_Takeover()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var receivedMessagesCount = 0;
+
+                await testEnvironment.StartServer();
+
+                // C1 will receive the last will!
+                var c1 = await testEnvironment.ConnectClient();
+                c1.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
+
+                // C2 has the last will defined.
+                var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithClientId("WillOwner");
+
+                var c2 = await testEnvironment.ConnectClient(clientOptions);
+
+                // C3 will do the connection takeover.
+                var c3 = await testEnvironment.ConnectClient(clientOptions);
+
+                await Task.Delay(1000);
+
+                Assert.AreEqual(0, receivedMessagesCount);
+            }
+        }
+
         async Task<IMqttClient> ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive)
         {
             try

Source/MQTTnet.Tests/Server/Will_Tests.cs

@@ -0,0 +1,71 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Implementations;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Tests.Server
+{
+    [TestClass]
+    public sealed class Will_Tests : BaseTestClass
+    {
+        [TestMethod]
+        public async Task Will_Message_Do_Not_Send_On_Clean_Disconnect()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                var receivedMessagesCount = 0;
+
+                await testEnvironment.StartServer();
+
+                var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will");
+
+                var c1 = await testEnvironment.ConnectClient();
+
+                c1.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
+
+                var c2 = await testEnvironment.ConnectClient(clientOptions);
+                await c2.DisconnectAsync().ConfigureAwait(false);
+
+                await Task.Delay(1000);
+
+                Assert.AreEqual(0, receivedMessagesCount);
+            }
+        }
+
+        [TestMethod]
+        public async Task Will_Message_Send()
+        {
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer();
+
+                var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder());
+
+                var receivedMessagesCount = 0;
+                c1.ApplicationMessageReceivedAsync += e =>
+                {
+                    Interlocked.Increment(ref receivedMessagesCount);
+                    return PlatformAbstractionLayer.CompletedTask;
+                };
+
+                await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
+
+                var clientOptions = new MqttClientOptionsBuilder().WithWillTopic("My/last/will").WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce);
+                var c2 = await testEnvironment.ConnectClient(clientOptions);
+                c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
+
+                await Task.Delay(1000);
+
+                Assert.AreEqual(1, receivedMessagesCount);
+            }
+        }
+    }
+}

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -264,8 +264,7 @@ namespace MQTTnet.Client
 
             return this;
         }
-
-        // TODO: Consider creating _MqttClientTcpOptionsBuilder_ as overload.
+        
         public MqttClientOptionsBuilder WithTcpServer(Action<MqttClientTcpOptions> optionsBuilder)
         {
             if (optionsBuilder == null)

Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs

@@ -2,6 +2,7 @@
 // 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.Net;
 using System.Net.Sockets;
 
 namespace MQTTnet.Client
@@ -10,8 +11,19 @@ namespace MQTTnet.Client
     {
         public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
 
+        /// <summary>
+        /// Gets the local endpoint (network card) which is used by the client.
+        /// Set it to _null_ to let the OS select the network card.
+        /// </summary>
+        public EndPoint LocalEndpoint { get; set; }
+        
         public int BufferSize { get; set; } = 8192;
 
+        /// <summary>
+        /// Gets or sets whether the underlying socket should run in dual mode.
+        /// Leaving this _null_ will avoid setting this value at socket level.
+        /// Setting this a value other than _null_ will throw an exception when only IPv4 is supported on the machine.
+        /// </summary>
         public bool? DualMode { get; set; }
 
         public LingerOption LingerState { get; set; } = new LingerOption(true, 0);

Source/MQTTnet/Client/Options/MqttClientWebSocketOptions.cs

@@ -7,23 +7,23 @@ using System.Net;
 
 namespace MQTTnet.Client
 {
-    public class MqttClientWebSocketOptions : IMqttClientChannelOptions
+    public sealed class MqttClientWebSocketOptions : IMqttClientChannelOptions
     {
-        public string Uri { get; set; }
+        public CookieContainer CookieContainer { get; set; }
+        
+        public MqttClientWebSocketProxyOptions ProxyOptions { get; set; }
 
         public IDictionary<string, string> RequestHeaders { get; set; }
 
         public ICollection<string> SubProtocols { get; set; } = new List<string> { "mqtt" };
 
-        public CookieContainer CookieContainer { get; set; }
-
-        public MqttClientWebSocketProxyOptions ProxyOptions { get; set; }
-
         public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
+        
+        public string Uri { get; set; }
 
         public override string ToString()
         {
             return Uri;
         }
     }
-}
+}

Source/MQTTnet/Client/Options/MqttClientWebSocketProxyOptions.cs

@@ -4,7 +4,7 @@
 
 namespace MQTTnet.Client
 {
-    public class MqttClientWebSocketProxyOptions
+    public sealed class MqttClientWebSocketProxyOptions
     {
         public string Address { get; set; }
 

Source/MQTTnet/Client/Receiving/MqttApplicationMessageReceivedEventArgs.cs

@@ -25,6 +25,7 @@ namespace MQTTnet.Client
             ClientId = clientId;
             ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
             PublishPacket = publishPacket;
+            PacketIdentifier = publishPacket.PacketIdentifier;
             _acknowledgeHandler = acknowledgeHandler;
         }
 
@@ -62,12 +63,17 @@ namespace MQTTnet.Client
         public bool IsHandled { get; set; }
 
         /// <summary>
-        /// Gets ir sets whether the library should send MQTT ACK packets automatically if required.
+        /// Gets or sets whether the library should send MQTT ACK packets automatically if required.
         /// </summary>
         public bool AutoAcknowledge { get; set; } = true;
         
         public object Tag { get; set; }
-        
+
+        /// <summary>
+        /// Gets the indentifier of MQTT packet
+        /// </summary>
+        public ushort PacketIdentifier { get; }
+
         public Task AcknowledgeAsync(CancellationToken cancellationToken) 
         {
             if (_acknowledgeHandler == null)

Source/MQTTnet/Implementations/MqttTcpChannel.cs

@@ -68,6 +68,11 @@ namespace MQTTnet.Implementations
                     socket = new CrossPlatformSocket(_tcpOptions.AddressFamily);
                 }
 
+                if (_tcpOptions.LocalEndpoint != null)
+                {
+                    socket.Bind(_tcpOptions.LocalEndpoint);
+                }
+                
                 socket.ReceiveBufferSize = _tcpOptions.BufferSize;
                 socket.SendBufferSize = _tcpOptions.BufferSize;
                 socket.SendTimeout = (int)_clientOptions.Timeout.TotalMilliseconds;

Source/MQTTnet/Implementations/MqttWebSocketChannel.cs

@@ -118,7 +118,6 @@ namespace MQTTnet.Implementations
 
         void SetupClientWebSocket(ClientWebSocket clientWebSocket)
         {
-
             if (_options.ProxyOptions != null)
             {
                 clientWebSocket.Options.Proxy = CreateProxy();

Source/MQTTnet/Internal/AsyncEventInvocator.cs

@@ -21,12 +21,14 @@ namespace MQTTnet.Internal
 
         public bool WrapsHandler(Action<TEventArgs> handler)
         {
-            return ReferenceEquals(_handler, handler);
+            // Do not use ReferenceEquals! It will not work with delegates.
+            return handler == _handler;
         }
         
         public bool WrapsHandler(Func<TEventArgs, Task> handler)
         {
-            return ReferenceEquals(_asyncHandler, handler);
+            // Do not use ReferenceEquals! It will not work with delegates.
+            return handler == _asyncHandler;
         }
         
         public Task InvokeAsync(TEventArgs eventArgs)

Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs

@@ -8,14 +8,20 @@ namespace MQTTnet.Server
 {
     public sealed class ApplicationMessageNotConsumedEventArgs : EventArgs
     {
+        public ApplicationMessageNotConsumedEventArgs(MqttApplicationMessage applicationMessage, string senderId)
+        {
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            SenderId = senderId;
+        }
+
         /// <summary>
         ///     Gets the application message which was not consumed by any client.
         /// </summary>
-        public MqttApplicationMessage ApplicationMessage { get; internal set; }
+        public MqttApplicationMessage ApplicationMessage { get; }
 
         /// <summary>
         ///     Gets the ID of the client which has sent the affected application message.
         /// </summary>
-        public string SenderId { get; internal set; }
+        public string SenderId { get; }
     }
 }

Source/MQTTnet/Server/Events/ClientAcknowledgedPublishPacketEventArgs.cs

@@ -0,0 +1,38 @@
+// 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;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public sealed class ClientAcknowledgedPublishPacketEventArgs : EventArgs
+    {
+        /// <summary>
+        ///     Gets the packet which was used for acknowledge. This can be a PubAck or PubComp packet.
+        /// </summary>
+        public MqttPacketWithIdentifier AcknowledgePacket { get; internal set; }
+
+        /// <summary>
+        ///     Gets the ID of the client which acknowledged a PUBLISH packet.
+        /// </summary>
+        public string ClientId { get; internal set; }
+
+        /// <summary>
+        ///     Gets whether the PUBLISH packet is fully acknowledged. This is the case for PUBACK (QoS 1) and PUBCOMP (QoS 2.
+        /// </summary>
+        public bool IsCompleted => AcknowledgePacket is MqttPubAckPacket || AcknowledgePacket is MqttPubCompPacket;
+
+        /// <summary>
+        ///     Gets the PUBLISH packet which was acknowledged.
+        /// </summary>
+        public MqttPublishPacket PublishPacket { get; internal set; }
+
+        /// <summary>
+        ///     Gets the session items which contain custom user data per session.
+        /// </summary>
+        public IDictionary SessionItems { get; internal set; }
+    }
+}

Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs

@@ -3,31 +3,46 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
 using MQTTnet.Formatter;
 
 namespace MQTTnet.Server
 {
     public sealed class ClientConnectedEventArgs : EventArgs
     {
-        /// <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; internal set; }
+        public ClientConnectedEventArgs(string clientId, string userName, MqttProtocolVersion protocolVersion, string endpoint, IDictionary sessionItems)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            UserName = userName;
+            ProtocolVersion = protocolVersion;
+            Endpoint = endpoint;
+            SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+        }
 
         /// <summary>
-        /// Gets the user name of the connected client.
+        ///     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 UserName { get; internal set; }
+        public string ClientId { get; }
 
         /// <summary>
-        /// Gets the protocol version which is used by the connected client.
+        ///     Gets the endpoint of the connected client.
         /// </summary>
-        public MqttProtocolVersion ProtocolVersion { get; internal set; }
+        public string Endpoint { get; }
 
         /// <summary>
-        /// Gets the endpoint of the connected client.
+        ///     Gets the protocol version which is used by the connected client.
         /// </summary>
-        public string Endpoint { get; internal set; }
+        public MqttProtocolVersion ProtocolVersion { get; }
+
+        /// <summary>
+        ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary SessionItems { get; }
+
+        /// <summary>
+        ///     Gets the user name of the connected client.
+        /// </summary>
+        public string UserName { get; }
     }
-}
+}

Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs

@@ -3,19 +3,33 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
 
 namespace MQTTnet.Server
 {
     public sealed class ClientDisconnectedEventArgs : EventArgs
     {
+        public ClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType, string endpoint, IDictionary sessionItems)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            DisconnectType = disconnectType;
+            Endpoint = endpoint;
+            SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+        }
+
         /// <summary>
-        /// Gets the client identifier.
-        /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
+        ///     Gets the client identifier.
+        ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        public string ClientId { get; internal set; }
+        public string ClientId { get; }
 
-        public MqttClientDisconnectType DisconnectType { get; internal set; }
+        public MqttClientDisconnectType DisconnectType { get; }
 
-        public string Endpoint { get; internal set; }
+        public string Endpoint { get; }
+
+        /// <summary>
+        ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary SessionItems { get; }
     }
-}
+}

Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs

@@ -3,22 +3,35 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Server
 {
     public sealed class ClientSubscribedTopicEventArgs : EventArgs
     {
-        /// <summary>
-        /// Gets the client identifier.
-        /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
-        /// </summary>
-        public string ClientId { get; internal set; }
+        public ClientSubscribedTopicEventArgs(string clientId, MqttTopicFilter topicFilter, IDictionary sessionItems)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
+            SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+        }
 
         /// <summary>
-        /// Gets the topic filter.
-        /// The topic filter can contain topics and wildcards.
+        ///     Gets the client identifier.
+        ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        public MqttTopicFilter TopicFilter { get; internal set; }
+        public string ClientId { get; }
+
+        /// <summary>
+        ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary SessionItems { get; }
+
+        /// <summary>
+        ///     Gets the topic filter.
+        ///     The topic filter can contain topics and wildcards.
+        /// </summary>
+        public MqttTopicFilter TopicFilter { get; }
     }
-}
+}

Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs

@@ -3,21 +3,34 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
 
 namespace MQTTnet.Server
 {
     public sealed class ClientUnsubscribedTopicEventArgs : EventArgs
     {
-        /// <summary>
-        /// Gets the client identifier.
-        /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
-        /// </summary>
-        public string ClientId { get; internal set; }
+        public ClientUnsubscribedTopicEventArgs(string clientId, string topicFilter, IDictionary sessionItems)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
+            SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+        }
 
         /// <summary>
-        /// Gets or sets the topic filter.
-        /// The topic filter can contain topics and wildcards.
+        ///     Gets the client identifier.
+        ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        public string TopicFilter { get; internal set; }
+        public string ClientId { get; }
+
+        /// <summary>
+        ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary SessionItems { get; }
+
+        /// <summary>
+        ///     Gets or sets the topic filter.
+        ///     The topic filter can contain topics and wildcards.
+        /// </summary>
+        public string TopicFilter { get; }
     }
-}
+}

Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
 using System.Threading;
 using MQTTnet.Packets;
 
@@ -10,30 +11,44 @@ namespace MQTTnet.Server
 {
     public sealed class InterceptingPacketEventArgs : EventArgs
     {
+        public InterceptingPacketEventArgs(CancellationToken cancellationToken, string clientId, string endpoint, MqttPacket packet, IDictionary sessionItems)
+        {
+            CancellationToken = cancellationToken;
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            Endpoint = endpoint;
+            Packet = packet ?? throw new ArgumentNullException(nameof(packet));
+            SessionItems = sessionItems;
+        }
+
         /// <summary>
-        /// Gets the client ID which has sent the packet or will receive the packet.
+        ///     Gets the cancellation token from the connection managing thread.
+        ///     Use this in further event processing.
         /// </summary>
-        public string ClientId { get; internal set; }
-        
+        public CancellationToken CancellationToken { get; }
+
         /// <summary>
-        /// Gets the endpoint of the sending or receiving client.
+        ///     Gets the client ID which has sent the packet or will receive the packet.
         /// </summary>
-        public string Endpoint { get; internal set; }
-        
+        public string ClientId { get; }
+
         /// <summary>
-        /// Gets or sets the MQTT packet which was received or will be sent.
+        ///     Gets the endpoint of the sending or receiving client.
+        /// </summary>
+        public string Endpoint { get; }
+
+        /// <summary>
+        ///     Gets or sets the MQTT packet which was received or will be sent.
         /// </summary>
         public MqttPacket Packet { get; set; }
 
         /// <summary>
-        /// Gets or sets whether the packet should be processed or not.
+        ///     Gets or sets whether the packet should be processed or not.
         /// </summary>
         public bool ProcessPacket { get; set; } = true;
-        
+
         /// <summary>
-        /// Gets the cancellation token from the connection managing thread.
-        /// Use this in further event processing.
+        ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
         /// </summary>
-        public CancellationToken CancellationToken { get; internal set; }
+        public IDictionary SessionItems { get; }
     }
 }

Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs

@@ -10,18 +10,26 @@ namespace MQTTnet.Server
 {
     public sealed class InterceptingPublishEventArgs : EventArgs
     {
+        public InterceptingPublishEventArgs(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken, string clientId, IDictionary sessionItems)
+        {
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            CancellationToken = cancellationToken;
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+        }
+
         public MqttApplicationMessage ApplicationMessage { get; set; }
 
         /// <summary>
         ///     Gets the cancellation token which can indicate that the client connection gets down.
         /// </summary>
-        public CancellationToken CancellationToken { get; internal set; }
+        public CancellationToken CancellationToken { get; }
 
         /// <summary>
         ///     Gets the client identifier.
         ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        public string ClientId { get; internal set; }
+        public string ClientId { get; }
 
         public bool CloseConnection { get; set; }
 
@@ -38,6 +46,6 @@ namespace MQTTnet.Server
         /// <summary>
         ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
         /// </summary>
-        public IDictionary SessionItems { get; internal set; }
+        public IDictionary SessionItems { get; }
     }
 }

Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs

@@ -12,16 +12,28 @@ namespace MQTTnet.Server
 {
     public sealed class InterceptingSubscriptionEventArgs : EventArgs
     {
+        public InterceptingSubscriptionEventArgs(
+            CancellationToken cancellationToken,
+            string clientId,
+            MqttSessionStatus session,
+            MqttTopicFilter topicFilter)
+        {
+            CancellationToken = cancellationToken;
+            ClientId = clientId;
+            Session = session;
+            TopicFilter = topicFilter;
+        }
+
         /// <summary>
         ///     Gets the cancellation token which can indicate that the client connection gets down.
         /// </summary>
-        public CancellationToken CancellationToken { get; internal set; }
+        public CancellationToken CancellationToken { get; }
 
         /// <summary>
         ///     Gets the client identifier.
         ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        public string ClientId { get; internal set; }
+        public string ClientId { get; }
 
         /// <summary>
         ///     Gets or sets whether the broker should close the client connection.
@@ -48,12 +60,12 @@ namespace MQTTnet.Server
         /// <summary>
         ///     Gets the current client session.
         /// </summary>
-        public MqttSessionStatus Session { get; internal set; }
+        public MqttSessionStatus Session { get; }
 
         /// <summary>
         ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
         /// </summary>
-        public IDictionary SessionItems { get; internal set; }
+        public IDictionary SessionItems => Session.Items;
 
         /// <summary>
         ///     Gets or sets the topic filter.

Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs

@@ -10,16 +10,24 @@ namespace MQTTnet.Server
 {
     public sealed class InterceptingUnsubscriptionEventArgs : EventArgs
     {
+        public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, IDictionary sessionItems, string topic)
+        {
+            CancellationToken = cancellationToken;
+            ClientId = clientId;
+            SessionItems = sessionItems;
+            Topic = topic;
+        }
+
         /// <summary>
         ///     Gets the cancellation token which can indicate that the client connection gets down.
         /// </summary>
-        public CancellationToken CancellationToken { get; internal set; }
+        public CancellationToken CancellationToken { get; }
 
         /// <summary>
         ///     Gets the client identifier.
         ///     Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
         /// </summary>
-        public string ClientId { get; internal set; }
+        public string ClientId { get; }
 
         /// <summary>
         ///     Gets or sets whether the broker should close the client connection.
@@ -41,7 +49,7 @@ namespace MQTTnet.Server
         /// <summary>
         ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
         /// </summary>
-        public IDictionary SessionItems { get; internal set; }
+        public IDictionary SessionItems { get; }
 
         /// <summary>
         ///     Gets or sets the MQTT topic.
@@ -50,6 +58,6 @@ namespace MQTTnet.Server
         ///     The topic consists of one or more topic levels. Each topic level is separated by a forward slash (topic level
         ///     separator).
         /// </summary>
-        public string Topic { get; internal set; }
+        public string Topic { get; }
     }
 }

Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs

@@ -9,10 +9,17 @@ namespace MQTTnet.Server
 {
     public sealed class RetainedMessageChangedEventArgs : EventArgs
     {
-        public string ClientId { get; internal set; }
-        
-        public MqttApplicationMessage ChangedRetainedMessage { get; internal set; }
-        
-        public List<MqttApplicationMessage> StoredRetainedMessages { get; internal set; }
+        public RetainedMessageChangedEventArgs(string clientId, MqttApplicationMessage changedRetainedMessage, List<MqttApplicationMessage> storedRetainedMessages)
+        {
+            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            ChangedRetainedMessage = changedRetainedMessage ?? throw new ArgumentNullException(nameof(changedRetainedMessage));
+            StoredRetainedMessages = storedRetainedMessages ?? throw new ArgumentNullException(nameof(storedRetainedMessages));
+        }
+
+        public MqttApplicationMessage ChangedRetainedMessage { get; }
+
+        public string ClientId { get; }
+
+        public List<MqttApplicationMessage> StoredRetainedMessages { get; }
     }
 }

Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs

@@ -3,14 +3,26 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
 
 namespace MQTTnet.Server
 {
     public sealed class SessionDeletedEventArgs : EventArgs
     {
+        public SessionDeletedEventArgs(string id, IDictionary sessionItems)
+        {
+            Id = id ?? throw new ArgumentNullException(nameof(id));
+            SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+        }
+
         /// <summary>
-        /// Gets the ID of the session.
+        ///     Gets the ID of the session.
         /// </summary>
-        public string Id { get; internal set; }
+        public string Id { get; }
+
+        /// <summary>
+        ///     Gets or sets a key/value collection that can be used to share data within the scope of this session.
+        /// </summary>
+        public IDictionary SessionItems { get; }
     }
 }

Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs

@@ -17,13 +17,12 @@ namespace MQTTnet.Server
 {
     public sealed class ValidatingConnectionEventArgs : EventArgs
     {
-        readonly IMqttChannelAdapter _clientAdapter;
         readonly MqttConnectPacket _connectPacket;
 
         public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
         {
             _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
-            _clientAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
+            ChannelAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
         }
 
         /// <summary>
@@ -44,6 +43,12 @@ namespace MQTTnet.Server
         /// </summary>
         public string AuthenticationMethod => _connectPacket.AuthenticationMethod;
 
+        /// <summary>
+        ///     Gets the channel adapter. This can be a _MqttConnectionContext_ (used in ASP.NET), a _MqttChannelAdapter_ (used for
+        ///     TCP or WebSockets) or a custom implementation.
+        /// </summary>
+        public IMqttChannelAdapter ChannelAdapter { get; }
+
         /// <summary>
         ///     Gets or sets a value indicating whether clean sessions are used or not.
         ///     When a client connects to a broker it can connect using either a non persistent connection (clean session) or a
@@ -56,7 +61,7 @@ namespace MQTTnet.Server
         /// </summary>
         public bool? CleanSession => _connectPacket.CleanSession;
 
-        public X509Certificate2 ClientCertificate => _clientAdapter.ClientCertificate;
+        public X509Certificate2 ClientCertificate => ChannelAdapter.ClientCertificate;
 
         /// <summary>
         ///     Gets the client identifier.
@@ -64,9 +69,9 @@ namespace MQTTnet.Server
         /// </summary>
         public string ClientId => _connectPacket.ClientId;
 
-        public string Endpoint => _clientAdapter.Endpoint;
+        public string Endpoint => ChannelAdapter.Endpoint;
 
-        public bool IsSecureConnection => _clientAdapter.IsSecureConnection;
+        public bool IsSecureConnection => ChannelAdapter.IsSecureConnection;
 
         /// <summary>
         ///     Gets or sets the keep alive period.
@@ -85,7 +90,7 @@ namespace MQTTnet.Server
 
         public string Password => Encoding.UTF8.GetString(RawPassword ?? PlatformAbstractionLayer.EmptyByteArray);
 
-        public MqttProtocolVersion ProtocolVersion => _clientAdapter.PacketFormatterAdapter.ProtocolVersion;
+        public MqttProtocolVersion ProtocolVersion => ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
 
         public byte[] RawPassword => _connectPacket.Password;
 
@@ -161,7 +166,10 @@ namespace MQTTnet.Server
         /// </summary>
         public ushort TopicAliasMaximum => _connectPacket.TopicAliasMaximum;
 
+        [Obsolete("This property name has a typo. Use 'UserName' instead. This one will be removed soon.")]
         public string Username => _connectPacket.Username;
+        
+        public string UserName => _connectPacket.Username;
 
         /// <summary>
         ///     Gets or sets the user properties.

Source/MQTTnet/Server/Internal/MqttClient.cs

@@ -11,6 +11,7 @@ using MQTTnet.Client;
 using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
+using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.PacketDispatcher;
 using MQTTnet.Packets;
@@ -20,20 +21,20 @@ namespace MQTTnet.Server
 {
     public sealed class MqttClient
     {
-        readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
+        readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
+        readonly MqttConnectPacket _connectPacket;
+        readonly MqttServerEventContainer _eventContainer;
+        readonly MqttNetSourceLogger _logger;
         readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
         readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
-        readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
-        
-        readonly MqttClientSessionsManager _sessionsManager;
-        readonly MqttNetSourceLogger _logger;
-   
+
         readonly MqttServerOptions _serverOptions;
-        readonly MqttServerEventContainer _eventContainer;
-        readonly MqttConnectPacket _connectPacket;
+
+        readonly MqttClientSessionsManager _sessionsManager;
+        readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
 
         CancellationTokenSource _cancellationToken;
-        
+
         public MqttClient(
             MqttConnectPacket connectPacket,
             IMqttChannelAdapter channelAdapter,
@@ -53,43 +54,31 @@ namespace MQTTnet.Server
             Session = session ?? throw new ArgumentNullException(nameof(session));
             _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
 
-            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            if (logger == null)
+            {
+                throw new ArgumentNullException(nameof(logger));
+            }
+
             _logger = logger.WithSource(nameof(MqttClient));
         }
 
-        public string Id => _connectPacket.ClientId;
+        public IMqttChannelAdapter ChannelAdapter { get; }
 
         public string Endpoint { get; }
 
-        public IMqttChannelAdapter ChannelAdapter { get; }
+        public string Id => _connectPacket.ClientId;
 
-        public MqttClientStatistics Statistics { get; } = new MqttClientStatistics();
+        public bool IsCleanDisconnect { get; private set; }
 
         public bool IsRunning { get; private set; }
 
+        public bool IsTakenOver { get; set; }
+
         public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod;
 
         public MqttSession Session { get; }
 
-        public bool IsTakenOver { get; set; }
-
-        public bool IsCleanDisconnect { get; private set; }
-
-        public async Task StopAsync(MqttDisconnectReasonCode reason)
-        {
-            IsRunning = false;
-
-            if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout)
-            {
-                // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
-                // token because the entire connection is closed (disposed) as soon as the cancellation
-                // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
-                // at the client!
-                await TrySendDisconnectPacket(reason).ConfigureAwait(false);
-            }
-
-            StopInternal();
-        }
+        public MqttClientStatistics Statistics { get; } = new MqttClientStatistics();
 
         public void ResetStatistics()
         {
@@ -110,7 +99,7 @@ namespace MQTTnet.Server
                 try
                 {
                     Task.Run(() => SendPacketsLoop(cancellationToken), cancellationToken).RunInBackground(_logger);
-                    
+
                     IsRunning = true;
 
                     await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
@@ -130,10 +119,10 @@ namespace MQTTnet.Server
             {
                 var willPublishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket);
                 var willApplicationMessage = _applicationMessageFactory.Create(willPublishPacket);
-                
-                _= _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage);
+
+                _ = _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage);
                 Session.WillMessageSent = true;
-                
+
                 _logger.Info("Client '{0}': Published will message.", Id);
             }
 
@@ -149,33 +138,233 @@ namespace MQTTnet.Server
                 // This might break the protocol but the user wants that.
                 return;
             }
-            
+
             await ChannelAdapter.SendPacketAsync(packet, cancellationToken).ConfigureAwait(false);
             Statistics.HandleSentPacket(packet);
         }
 
+        public async Task StopAsync(MqttDisconnectReasonCode reason)
+        {
+            IsRunning = false;
+
+            if (reason == MqttDisconnectReasonCode.SessionTakenOver || reason == MqttDisconnectReasonCode.KeepAliveTimeout)
+            {
+                // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
+                // token because the entire connection is closed (disposed) as soon as the cancellation
+                // token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
+                // at the client!
+                await TrySendDisconnectPacket(reason).ConfigureAwait(false);
+            }
+
+            StopInternal();
+        }
+
+        Task ClientAcknowledgedPublishPacket(MqttPublishPacket publishPacket, MqttPacketWithIdentifier acknowledgePacket)
+        {
+            if (_eventContainer.ClientAcknowledgedPublishPacketEvent.HasHandlers)
+            {
+                var eventArgs = new ClientAcknowledgedPublishPacketEventArgs
+                {
+                    PublishPacket = publishPacket,
+                    AcknowledgePacket = acknowledgePacket,
+                    ClientId = Id,
+                    SessionItems = Session.Items
+                };
+
+                return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger);
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
+        Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket)
+        {
+            var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
+
+            if (acknowledgedPublishPacket != null)
+            {
+                return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubAckPacket);
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
+        Task HandleIncomingPubCompPacket(MqttPubCompPacket pubCompPacket)
+        {
+            var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier);
+
+            if (acknowledgedPublishPacket != null)
+            {
+                return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubCompPacket);
+            }
+
+            return PlatformAbstractionLayer.CompletedTask;
+        }
+
+        async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
+        {
+            HandleTopicAlias(publishPacket);
+
+            InterceptingPublishEventArgs interceptingPublishEventArgs = null;
+            var applicationMessage = _applicationMessageFactory.Create(publishPacket);
+            var closeConnection = false;
+            var processPublish = true;
+
+            if (_eventContainer.InterceptingPublishEvent.HasHandlers)
+            {
+                interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, Id, Session.Items);
+                if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
+                {
+                    // This can happen if a topic alias us used but the topic is
+                    // unknown to the server.
+                    interceptingPublishEventArgs.Response.ReasonCode = MqttPubAckReasonCode.TopicNameInvalid;
+                    interceptingPublishEventArgs.ProcessPublish = false;
+                }
+
+                await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
+
+                applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
+                closeConnection = interceptingPublishEventArgs.CloseConnection;
+                processPublish = interceptingPublishEventArgs.ProcessPublish;
+            }
+
+            if (closeConnection)
+            {
+                await StopAsync(MqttDisconnectReasonCode.UnspecifiedError);
+                return;
+            }
+
+            if (processPublish && applicationMessage != null)
+            {
+                await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false);
+            }
+
+            switch (publishPacket.QualityOfServiceLevel)
+            {
+                case MqttQualityOfServiceLevel.AtMostOnce:
+                {
+                    // Do nothing since QoS 0 has no ACK at all!
+                    break;
+                }
+                case MqttQualityOfServiceLevel.AtLeastOnce:
+                {
+                    var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs);
+                    Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
+                    break;
+                }
+                case MqttQualityOfServiceLevel.ExactlyOnce:
+                {
+                    var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs);
+                    Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
+                    break;
+                }
+                default:
+                {
+                    throw new MqttCommunicationException("Received a not supported QoS level.");
+                }
+            }
+        }
+
+        async Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
+        {
+            var acknowledgedPublishPacket = Session.PeekAcknowledgePublishPacket(pubRecPacket.PacketIdentifier);
+
+            if (acknowledgedPublishPacket != null)
+            {
+                await ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubRecPacket).ConfigureAwait(false);
+            }
+
+            var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
+            Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
+        }
+
+        void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
+        {
+            var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
+            Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket));
+        }
+
+        async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
+        {
+            var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
+
+            var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
+
+            Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket));
+
+            if (subscribeResult.CloseConnection)
+            {
+                StopInternal();
+                return;
+            }
+
+            if (subscribeResult.RetainedMessages != null)
+            {
+                foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
+                {
+                    var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
+                    Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
+                }
+            }
+        }
+
+
+        async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
+        {
+            var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
+
+            var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
+
+            Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket));
+
+            if (unsubscribeResult.CloseConnection)
+            {
+                StopInternal();
+            }
+        }
+
+        void HandleTopicAlias(MqttPublishPacket publishPacket)
+        {
+            if (publishPacket.TopicAlias == 0)
+            {
+                return;
+            }
+
+            lock (_topicAlias)
+            {
+                if (!string.IsNullOrEmpty(publishPacket.Topic))
+                {
+                    _topicAlias[publishPacket.TopicAlias] = publishPacket.Topic;
+                }
+                else
+                {
+                    if (_topicAlias.TryGetValue(publishPacket.TopicAlias, out var topic))
+                    {
+                        publishPacket.Topic = topic;
+                    }
+                    else
+                    {
+                        _logger.Warning("Client '{0}': Received invalid topic alias ({1}).", Id, publishPacket.TopicAlias);
+                    }
+                }
+            }
+        }
+
         async Task<MqttPacket> InterceptPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
         {
             if (!_eventContainer.InterceptingOutboundPacketEvent.HasHandlers)
             {
                 return packet;
             }
-            
-            var interceptingPacketEventArgs = new InterceptingPacketEventArgs
-            {
-                ClientId = Id,
-                Endpoint = Endpoint,
-                Packet = packet,
-                CancellationToken = cancellationToken
-            };
-            
+
+            var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items);
             await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
-                
+
             if (!interceptingPacketEventArgs.ProcessPacket || packet == null)
             {
                 return null;
             }
-                
+
             return interceptingPacketEventArgs.Packet;
         }
 
@@ -189,7 +378,7 @@ namespace MQTTnet.Server
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     await Task.Yield();
-                    
+
                     var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
                     if (packet == null)
                     {
@@ -197,22 +386,15 @@ namespace MQTTnet.Server
                     }
 
                     var processPacket = true;
-                    
+
                     if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
                     {
-                        var interceptingPacketEventArgs = new InterceptingPacketEventArgs
-                        {
-                            ClientId = Id,
-                            Endpoint = Endpoint,
-                            Packet = packet,
-                            CancellationToken = cancellationToken
-                        };
-                    
+                        var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items);
                         await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
                         packet = interceptingPacketEventArgs.Packet;
                         processPacket = interceptingPacketEventArgs.ProcessPacket;
                     }
-                    
+
                     if (!processPacket || packet == null)
                     {
                         // Restart the receiving process to get the next packet ignoring the current one..
@@ -227,15 +409,15 @@ namespace MQTTnet.Server
                     }
                     else if (packet is MqttPubAckPacket pubAckPacket)
                     {
-                        Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
+                        await HandleIncomingPubAckPacket(pubAckPacket).ConfigureAwait(false);
                     }
                     else if (packet is MqttPubCompPacket pubCompPacket)
                     {
-                        Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier);
+                        await HandleIncomingPubCompPacket(pubCompPacket).ConfigureAwait(false);
                     }
                     else if (packet is MqttPubRecPacket pubRecPacket)
                     {
-                        HandleIncomingPubRecPacket(pubRecPacket);
+                        await HandleIncomingPubRecPacket(pubRecPacket).ConfigureAwait(false);
                     }
                     else if (packet is MqttPubRelPacket pubRelPacket)
                     {
@@ -353,7 +535,7 @@ namespace MQTTnet.Server
                 StopInternal();
             }
         }
-        
+
         void StopInternal()
         {
             try
@@ -367,162 +549,12 @@ namespace MQTTnet.Server
             }
         }
 
-        void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
-        {
-            var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket));
-        }
-
-        void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
-        {
-            var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket));
-        }
-
-        async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
-        {
-            var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
-
-            var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
-
-            Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket));
-
-            if (subscribeResult.CloseConnection)
-            {
-                StopInternal();
-                return;
-            }
-
-            if (subscribeResult.RetainedMessages != null)
-            {
-                foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages)
-                {
-                    var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
-                    Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket));
-                }
-            }
-        }
-
-        async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
-        {
-            var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
-
-            var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
-
-            Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket));
-
-            if (unsubscribeResult.CloseConnection)
-            {
-                StopInternal();
-            }
-        }
-       
-        async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
-        {
-            HandleTopicAlias(publishPacket);
-
-            InterceptingPublishEventArgs interceptingPublishEventArgs = null;
-            var applicationMessage = _applicationMessageFactory.Create(publishPacket);
-            var closeConnection = false;
-            var processPublish = true;
-            
-            if (_eventContainer.InterceptingPublishEvent.HasHandlers)
-            {
-                interceptingPublishEventArgs = new InterceptingPublishEventArgs
-                {
-                    ClientId = Id,
-                    ApplicationMessage = applicationMessage,
-                    SessionItems = Session.Items,
-                    ProcessPublish = true,
-                    CloseConnection = false,
-                    CancellationToken = cancellationToken
-                };
-
-                if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
-                {
-                    // This can happen if a topic alias us used but the topic is
-                    // unknown to the server.
-                    interceptingPublishEventArgs.Response.ReasonCode = MqttPubAckReasonCode.TopicNameInvalid;
-                    interceptingPublishEventArgs.ProcessPublish = false;
-                }
-                
-                await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
-                
-                applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
-                closeConnection = interceptingPublishEventArgs.CloseConnection;
-                processPublish = interceptingPublishEventArgs.ProcessPublish;
-            }
-
-            if (closeConnection)
-            {
-                await StopAsync(MqttDisconnectReasonCode.UnspecifiedError);
-                return;
-            }
-
-            if (processPublish && applicationMessage != null)
-            {
-                await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false);    
-            }
-            
-            switch (publishPacket.QualityOfServiceLevel)
-            {
-                case MqttQualityOfServiceLevel.AtMostOnce:
-                {
-                    // Do nothing since QoS 0 has no ack at all!
-                    break;
-                }
-                case MqttQualityOfServiceLevel.AtLeastOnce:
-                {
-                    var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs);
-                    Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
-                    break;
-                }
-                case MqttQualityOfServiceLevel.ExactlyOnce:
-                {
-                    var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs);
-                    Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
-                    break;
-                }
-                default:
-                {
-                    throw new MqttCommunicationException("Received a not supported QoS level.");
-                }
-            }
-        }
-
-        void HandleTopicAlias(MqttPublishPacket publishPacket)
-        {
-            if (publishPacket.TopicAlias == 0)
-            {
-                return;
-            }
-            
-            lock (_topicAlias)
-            {
-                if (!string.IsNullOrEmpty(publishPacket.Topic))
-                {
-                    _topicAlias[publishPacket.TopicAlias] = publishPacket.Topic;
-                }
-                else
-                {
-                    if (_topicAlias.TryGetValue(publishPacket.TopicAlias, out var topic))
-                    {
-                        publishPacket.Topic = topic;
-                    }
-                    else
-                    {
-                        _logger.Warning("Client '{0}': Received invalid topic alias ({1}).", Id, publishPacket.TopicAlias);
-                    }
-                }
-            }
-        }
-
         async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode)
         {
             try
             {
                 var disconnectPacket = _packetFactories.Disconnect.Create(reasonCode);
-                
+
                 using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
                 {
                     await SendPacketAsync(disconnectPacket, timeout.Token).ConfigureAwait(false);

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -109,13 +109,9 @@ namespace MQTTnet.Server
 
             try
             {
-                if (_eventContainer.SessionDeletedEvent.HasHandlers)
+                if (_eventContainer.SessionDeletedEvent.HasHandlers && session != null)
                 {
-                    var eventArgs = new SessionDeletedEventArgs
-                    {
-                        Id = session?.Id
-                    };
-
+                    var eventArgs = new SessionDeletedEventArgs(clientId, session.Items);
                     await _eventContainer.SessionDeletedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
                 }
             }
@@ -295,13 +291,12 @@ namespace MQTTnet.Server
 
                 if (_eventContainer.ClientConnectedEvent.HasHandlers)
                 {
-                    var eventArgs = new ClientConnectedEventArgs
-                    {
-                        ClientId = connectPacket.ClientId,
-                        UserName = connectPacket.Username,
-                        ProtocolVersion = channelAdapter.PacketFormatterAdapter.ProtocolVersion,
-                        Endpoint = channelAdapter.Endpoint
-                    };
+                    var eventArgs = new ClientConnectedEventArgs(
+                        connectPacket.ClientId,
+                        connectPacket.Username,
+                        channelAdapter.PacketFormatterAdapter.ProtocolVersion,
+                        channelAdapter.Endpoint,
+                        client.Session.Items);
 
                     await _eventContainer.ClientConnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                 }
@@ -340,12 +335,8 @@ namespace MQTTnet.Server
 
                     if (client.Id != null && !client.IsTakenOver && _eventContainer.ClientDisconnectedEvent.HasHandlers)
                     {
-                        var eventArgs = new ClientDisconnectedEventArgs
-                        {
-                            ClientId = client.Id,
-                            DisconnectType = client.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean,
-                            Endpoint = endpoint
-                        };
+                        var disconnectType = client.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean;
+                        var eventArgs = new ClientDisconnectedEventArgs(client.Id, disconnectType, endpoint, client.Session.Items);
 
                         await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                     }
@@ -533,13 +524,7 @@ namespace MQTTnet.Server
 
                     if (_eventContainer.ClientConnectedEvent.HasHandlers)
                     {
-                        var eventArgs = new ClientDisconnectedEventArgs
-                        {
-                            ClientId = existing.Id,
-                            DisconnectType = MqttClientDisconnectType.Takeover,
-                            Endpoint = existing.Endpoint
-                        };
-
+                        var eventArgs = new ClientDisconnectedEventArgs(existing.Id, MqttClientDisconnectType.Takeover, existing.Endpoint, existing.Session.Items);
                         await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                     }
                 }
@@ -572,12 +557,7 @@ namespace MQTTnet.Server
                 return;
             }
 
-            var eventArgs = new ApplicationMessageNotConsumedEventArgs
-            {
-                ApplicationMessage = applicationMessage,
-                SenderId = senderId
-            };
-
+            var eventArgs = new ApplicationMessageNotConsumedEventArgs(applicationMessage, senderId);
             await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
         }
 
@@ -623,25 +603,25 @@ namespace MQTTnet.Server
 
         async Task<ValidatingConnectionEventArgs> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
         {
-            var context = new ValidatingConnectionEventArgs(connectPacket, channelAdapter)
+            var eventArgs = new ValidatingConnectionEventArgs(connectPacket, channelAdapter)
             {
                 SessionItems = new ConcurrentDictionary<object, object>()
             };
 
-            await _eventContainer.ValidatingConnectionEvent.InvokeAsync(context).ConfigureAwait(false);
+            await _eventContainer.ValidatingConnectionEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
 
             // Check the client ID and set a random one if supported.
             if (string.IsNullOrEmpty(connectPacket.ClientId) && channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
             {
-                connectPacket.ClientId = context.AssignedClientIdentifier;
+                connectPacket.ClientId = eventArgs.AssignedClientIdentifier;
             }
 
             if (string.IsNullOrEmpty(connectPacket.ClientId))
             {
-                context.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
+                eventArgs.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
             }
 
-            return context;
+            return eventArgs;
         }
     }
 }

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -217,12 +217,7 @@ namespace MQTTnet.Server
             {
                 foreach (var finalTopicFilter in finalTopicFilters)
                 {
-                    var eventArgs = new ClientSubscribedTopicEventArgs
-                    {
-                        ClientId = _session.Id,
-                        TopicFilter = finalTopicFilter
-                    };
-
+                    var eventArgs = new ClientSubscribedTopicEventArgs(_session.Id, finalTopicFilter, _session.Items);
                     await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                 }
             }
@@ -312,12 +307,7 @@ namespace MQTTnet.Server
             {
                 foreach (var topicFilter in unsubscribePacket.TopicFilters)
                 {
-                    var eventArgs = new ClientUnsubscribedTopicEventArgs
-                    {
-                        ClientId = _session.Id,
-                        TopicFilter = topicFilter
-                    };
-
+                    var eventArgs = new ClientUnsubscribedTopicEventArgs(_session.Id, topicFilter, _session.Items);
                     await _eventContainer.ClientUnsubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                 }
             }
@@ -472,14 +462,7 @@ namespace MQTTnet.Server
 
         async Task<InterceptingSubscriptionEventArgs> InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
         {
-            var eventArgs = new InterceptingSubscriptionEventArgs
-            {
-                ClientId = _session.Id,
-                TopicFilter = topicFilter,
-                SessionItems = _session.Items,
-                Session = new MqttSessionStatus(_session),
-                CancellationToken = cancellationToken
-            };
+            var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, new MqttSessionStatus(_session), topicFilter);
 
             if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
@@ -508,13 +491,7 @@ namespace MQTTnet.Server
 
         async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
         {
-            var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs
-            {
-                ClientId = _session.Id,
-                Topic = topicFilter,
-                SessionItems = _session.Items,
-                CancellationToken = cancellationToken
-            };
+            var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter);
 
             if (mqttSubscription == null)
             {

Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs

@@ -105,13 +105,7 @@ namespace MQTTnet.Server
                 {
                     using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
                     {
-                        var eventArgs = new RetainedMessageChangedEventArgs
-                        {
-                            ClientId = clientId,
-                            ChangedRetainedMessage = applicationMessage,
-                            StoredRetainedMessages = messagesForSave
-                        };
-                        
+                        var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave);
                         await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
                     }
                 }

Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs

@@ -9,40 +9,42 @@ namespace MQTTnet.Server
 {
     public sealed class MqttServerEventContainer
     {
-        public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
-        
-        public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
-        
+        public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<ApplicationMessageNotConsumedEventArgs>();
+
+        public AsyncEvent<ClientAcknowledgedPublishPacketEventArgs> ClientAcknowledgedPublishPacketEvent { get; } = new AsyncEvent<ClientAcknowledgedPublishPacketEventArgs>();
+
+        public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<ClientConnectedEventArgs>();
+
+        public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<ClientDisconnectedEventArgs>();
+
+        public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<ClientSubscribedTopicEventArgs>();
+
+        public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
+
+        public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
+
+        public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
+
         public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new AsyncEvent<InterceptingPublishEventArgs>();
         
-        public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new AsyncEvent<ValidatingConnectionEventArgs>();
-        
-        public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<ClientConnectedEventArgs>();
-        
-        public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<ClientDisconnectedEventArgs>();
-        
-        public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<ClientSubscribedTopicEventArgs>();
-        
-        public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
-        
-        public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new AsyncEvent<EventArgs>();
-        
-        public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new AsyncEvent<SessionDeletedEventArgs>();
-        
-        public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<ApplicationMessageNotConsumedEventArgs>();
-        
-        public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
-        
+        public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
+
+        public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
+
         public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new AsyncEvent<LoadingRetainedMessagesEventArgs>();
-        
+
+        public AsyncEvent<EventArgs> PreparingSessionEvent { get; } = new AsyncEvent<EventArgs>();
+
+        public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
+
         public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new AsyncEvent<EventArgs>();
-        
-        public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
-        
-        public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
-        
+
+        public AsyncEvent<SessionDeletedEventArgs> SessionDeletedEvent { get; } = new AsyncEvent<SessionDeletedEventArgs>();
+
         public AsyncEvent<EventArgs> StartedEvent { get; } = new AsyncEvent<EventArgs>();
-        
+
         public AsyncEvent<EventArgs> StoppedEvent { get; } = new AsyncEvent<EventArgs>();
+
+        public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new AsyncEvent<ValidatingConnectionEventArgs>();
     }
-}
+}

Source/MQTTnet/Server/Internal/MqttSession.cs

@@ -71,9 +71,18 @@ namespace MQTTnet.Server
 
         public bool WillMessageSent { get; set; }
 
-        public void AcknowledgePublishPacket(ushort packetIdentifier)
+        public MqttPublishPacket PeekAcknowledgePublishPacket(ushort packetIdentifier)
         {
-            _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out _);
+            // This will only return the matching PUBLISH packet but does not remove it.
+            // This is required for QoS 2.
+            _unacknowledgedPublishPackets.TryGetValue(packetIdentifier, out var publishPacket);
+            return publishPacket;
+        }
+        
+        public MqttPublishPacket AcknowledgePublishPacket(ushort packetIdentifier)
+        {
+            _unacknowledgedPublishPackets.TryRemove(packetIdentifier, out var publishPacket);
+            return publishPacket;
         }
 
         public void AddSubscribedTopic(string topic)

Source/MQTTnet/Server/MqttServer.cs

@@ -3,6 +3,8 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -19,17 +21,18 @@ namespace MQTTnet.Server
     public class MqttServer : Disposable
     {
         readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer();
-        
+
+        readonly IDictionary _sessionItems = new ConcurrentDictionary<object, object>();
         readonly ICollection<IMqttServerAdapter> _adapters;
+        readonly MqttClientSessionsManager _clientSessionsManager;
+        readonly MqttServerKeepAliveMonitor _keepAliveMonitor;
         readonly MqttNetSourceLogger _logger;
         readonly MqttServerOptions _options;
-        readonly IMqttNetLogger _rootLogger;
         readonly MqttRetainedMessagesManager _retainedMessagesManager;
-        readonly MqttServerKeepAliveMonitor _keepAliveMonitor;
-        readonly MqttClientSessionsManager _clientSessionsManager;
-        
+        readonly IMqttNetLogger _rootLogger;
+
         CancellationTokenSource _cancellationTokenSource;
-        
+
         public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -55,6 +58,12 @@ namespace MQTTnet.Server
             remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value);
         }
 
+        public event Func<ClientAcknowledgedPublishPacketEventArgs, Task> ClientAcknowledgedPublishPacketAsync
+        {
+            add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value);
+            remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value);
+        }
+
         public event Func<ClientConnectedEventArgs, Task> ClientConnectedAsync
         {
             add => _eventContainer.ClientConnectedEvent.AddHandler(value);
@@ -158,7 +167,7 @@ namespace MQTTnet.Server
         }
 
         public bool IsStarted => _cancellationTokenSource != null;
-        
+
         public Task DeleteRetainedMessagesAsync()
         {
             ThrowIfNotStarted();
@@ -174,10 +183,10 @@ namespace MQTTnet.Server
             }
 
             ThrowIfNotStarted();
-            
+
             return _clientSessionsManager.GetClient(id).StopAsync(reasonCode);
         }
-        
+
         public Task<IList<MqttClientStatus>> GetClientsAsync()
         {
             ThrowIfNotStarted();
@@ -199,7 +208,7 @@ namespace MQTTnet.Server
             return _clientSessionsManager.GetSessionStatusAsync();
         }
 
-        public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage)
+        public async Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage)
         {
             if (injectedApplicationMessage == null)
             {
@@ -215,7 +224,34 @@ namespace MQTTnet.Server
 
             ThrowIfNotStarted();
 
-            return _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, injectedApplicationMessage.ApplicationMessage);
+            var processPublish = true;
+            var applicationMessage = injectedApplicationMessage.ApplicationMessage;
+
+            if (_eventContainer.InterceptingPublishEvent.HasHandlers)
+            {
+                var interceptingPublishEventArgs = new InterceptingPublishEventArgs(
+                    applicationMessage,
+                    _cancellationTokenSource.Token,
+                    injectedApplicationMessage.SenderClientId,
+                    _sessionItems);
+
+                await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
+
+                applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
+                processPublish = interceptingPublishEventArgs.ProcessPublish;
+            }
+
+            if (!processPublish)
+            {
+                return;
+            }
+            
+            if (string.IsNullOrEmpty(applicationMessage.Topic))
+            {
+                throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported.");
+            }
+            
+            await _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, applicationMessage).ConfigureAwait(false);
         }
 
         public async Task StartAsync()
@@ -236,7 +272,7 @@ namespace MQTTnet.Server
             }
 
             await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
-            
+
             _logger.Info("Started.");
         }
 
@@ -266,7 +302,7 @@ namespace MQTTnet.Server
             }
 
             await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
-            
+
             _logger.Info("Stopped.");
         }
 
@@ -310,7 +346,7 @@ namespace MQTTnet.Server
 
             return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
         }
-        
+
         public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage)
         {
             if (retainedMessage == null)
@@ -347,7 +383,7 @@ namespace MQTTnet.Server
         void ThrowIfNotStarted()
         {
             ThrowIfDisposed();
-            
+
             if (_cancellationTokenSource == null)
             {
                 throw new InvalidOperationException("The MQTT server is not started.");
@@ -357,7 +393,7 @@ namespace MQTTnet.Server
         void ThrowIfStarted()
         {
             ThrowIfDisposed();
-            
+
             if (_cancellationTokenSource != null)
             {
                 throw new InvalidOperationException("The MQTT server is already started.");