Squashed commit of the following: 487-any-way-to-generate-events-on-receipt-of-puback-or-pubrec-packets

commit 3f2f8d4b0a31b20734eca3fff86c43bd1ac3e3ad
Merge: 34babbf2 d71ff3d6
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Sat Jul 23 11:06:22 2022 +0200

    Merge branch 'master' into 487-any-way-to-generate-events-on-receipt-of-puback-or-pubrec-packets

commit 34babbf2714b742bc4f3802804a1a07159d38e22
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Mon Jul 11 22:56:01 2022 +0200

    Fix typo.

commit add6277458f528fc03bf4a1ba228a55c5a8d1a80
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Sat Jul 2 12:50:13 2022 +0200

    Add new event for client acknowledges.

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;

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/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/ValidatingConnectionEventArgs.cs

@@ -161,7 +161,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,18 +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
+                {
+                    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.");
+                }
+            }
+        }
+
+        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,
@@ -168,14 +372,14 @@ namespace MQTTnet.Server
                 Packet = packet,
                 CancellationToken = cancellationToken
             };
-            
+
             await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
-                
+
             if (!interceptingPacketEventArgs.ProcessPacket || packet == null)
             {
                 return null;
             }
-                
+
             return interceptingPacketEventArgs.Packet;
         }
 
@@ -197,7 +401,7 @@ namespace MQTTnet.Server
                     }
 
                     var processPacket = true;
-                    
+
                     if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers)
                     {
                         var interceptingPacketEventArgs = new InterceptingPacketEventArgs
@@ -207,12 +411,12 @@ namespace MQTTnet.Server
                             Packet = packet,
                             CancellationToken = cancellationToken
                         };
-                    
+
                         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 +431,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 +557,7 @@ namespace MQTTnet.Server
                 StopInternal();
             }
         }
-        
+
         void StopInternal()
         {
             try
@@ -367,162 +571,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/MqttServerEventContainer.cs

@@ -9,40 +9,41 @@ 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<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<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new AsyncEvent<LoadingRetainedMessagesEventArgs>();
-        
-        public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new AsyncEvent<EventArgs>();
-        
+
+        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<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<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

@@ -60,6 +60,12 @@ namespace MQTTnet.Server
             add => _eventContainer.ClientConnectedEvent.AddHandler(value);
             remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value);
         }
+        
+        public event Func<ClientAcknowledgedPublishPacketEventArgs, Task> ClientAcknowledgedPublishPacketAsync
+        {
+            add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value);
+            remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value);
+        }
 
         public event Func<ClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
         {