Refactor code.

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -84,7 +84,7 @@ namespace MQTTnet.AspNetCore
             _output = Connection.Transport.Output;
         }
 
-        public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        public Task DisconnectAsync(CancellationToken cancellationToken)
         {
             _input?.Complete();
             _output?.Complete();

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs

@@ -205,25 +205,36 @@ namespace MQTTnet.Extensions.WebSocket4Net
                 _webSocket.Open();
 #pragma warning restore AsyncFixer02 // Long-running or blocking operations inside an async method
 
-                var exception = await MqttTaskTimeout.WaitAsync(c =>
+                using (var timeoutCts = new CancellationTokenSource(_clientOptions.CommunicationTimeout))
+                using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken))
                 {
-                    c.Register(() => taskCompletionSource.TrySetCanceled());
+                    using (linkedCts.Token.Register(() => taskCompletionSource.TrySetCanceled()))
-                    return taskCompletionSource.Task;
-                }, _clientOptions.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
-
-                if (exception != null)
-                {
-                    if (exception is AuthenticationException authenticationException)
                     {
-                        throw new MqttCommunicationException(authenticationException.InnerException);
+                        try
-                    }
+                        {
+                            await taskCompletionSource.Task.ConfigureAwait(false);
+                        }
+                        catch (Exception exception)
+                        {
+                            var timeoutReached = timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested;
+                            if (timeoutReached)
+                            {
+                                throw new MqttCommunicationTimedOutException(exception);
+                            }
+                            
+                            if (exception is AuthenticationException authenticationException)
+                            {
+                                throw new MqttCommunicationException(authenticationException.InnerException);
+                            }
 
-                    if (exception is OperationCanceledException)
+                            if (exception is OperationCanceledException)
-                    {
+                            {
-                        throw new MqttCommunicationTimedOutException();
+                                throw new MqttCommunicationTimedOutException();
-                    }
+                            }
 
-                    throw new MqttCommunicationException(exception);
+                            throw new MqttCommunicationException(exception);
+                        }
+                    }
                 }
             }
             catch (OperationCanceledException)

Source/MQTTnet/Adapter/IMqttChannelAdapter.cs

@@ -25,7 +25,7 @@ namespace MQTTnet.Adapter
 
         Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken);
 
-        Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken);
+        Task DisconnectAsync(CancellationToken cancellationToken);
 
         Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken);
 

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -76,40 +76,28 @@ namespace MQTTnet.Adapter
             }
             catch (Exception exception)
             {
-                if (IsWrappedException(exception))
+                if (!WrapAndThrowException(exception))
                 {
                     throw;
                 }
-
-                WrapAndThrowException(exception);
             }
         }
 
-        public async Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        public async Task DisconnectAsync(CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
             ThrowIfDisposed();
 
             try
             {
-                if (timeout == TimeSpan.Zero)
+                await _channel.DisconnectAsync(cancellationToken).ConfigureAwait(false);
-                {
-                    await _channel.DisconnectAsync(cancellationToken).ConfigureAwait(false);
-                }
-                else
-                {
-                    await MqttTaskTimeout.WaitAsync(
-                        t => _channel.DisconnectAsync(t), timeout, cancellationToken).ConfigureAwait(false);
-                }
             }
             catch (Exception exception)
             {
-                if (IsWrappedException(exception))
+                if (!WrapAndThrowException(exception))
                 {
                     throw;
                 }
-
-                WrapAndThrowException(exception);
             }
         }
 
@@ -136,12 +124,10 @@ namespace MQTTnet.Adapter
                 }
                 catch (Exception exception)
                 {
-                    if (IsWrappedException(exception))
+                    if (!WrapAndThrowException(exception))
                     {
                         throw;
                     }
-
-                    WrapAndThrowException(exception);
                 }
                 finally
                 {
@@ -192,12 +178,10 @@ namespace MQTTnet.Adapter
             }
             catch (Exception exception)
             {
-                if (IsWrappedException(exception))
+                if (!WrapAndThrowException(exception))
                 {
                     throw;
                 }
-
-                WrapAndThrowException(exception);
             }
 
             return null;
@@ -386,17 +370,17 @@ namespace MQTTnet.Adapter
 
             return value;
         }
-
+        
-        static bool IsWrappedException(Exception exception)
+        static bool WrapAndThrowException(Exception exception)
-        {
-            return exception is OperationCanceledException ||
-                   exception is MqttCommunicationTimedOutException ||
-                   exception is MqttCommunicationException ||
-                   exception is MqttProtocolViolationException;
-        }
-
-        static void WrapAndThrowException(Exception exception)
         {
+            if (exception is OperationCanceledException ||
+                exception is MqttCommunicationTimedOutException ||
+                exception is MqttCommunicationException ||
+                exception is MqttProtocolViolationException)
+            {
+                return false;
+            }
+            
             if (exception is IOException && exception.InnerException is SocketException innerException)
             {
                 exception = innerException;

Source/MQTTnet/Client/MqttClient.cs

@@ -23,6 +23,7 @@ namespace MQTTnet.Client
 {
     public class MqttClient : Disposable, IMqttClient
     {
+        readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
         readonly object _disconnectLock = new object();
@@ -156,8 +157,7 @@ namespace MQTTnet.Client
 
                 if (clientWasConnected)
                 {
-                    var disconnectPacketFactory = new MqttDisconnectPacketFactory();
+                    var disconnectPacket = _packetFactories.Disconnect.Create(options);
-                    var disconnectPacket = disconnectPacketFactory.Create(options);
                     await SendAsync(disconnectPacket, cancellationToken).ConfigureAwait(false);
                 }
             }
@@ -207,8 +207,7 @@ namespace MQTTnet.Client
             ThrowIfDisposed();
             ThrowIfNotConnected();
 
-            var subscribePacketFactory = new MqttSubscribePacketFactory();
+            var subscribePacket = _packetFactories.Subscribe.Create(options);
-            var subscribePacket = subscribePacketFactory.Create(options);
             subscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
 
             var subAckPacket = await SendAndReceiveAsync<MqttSubAckPacket>(subscribePacket, cancellationToken).ConfigureAwait(false);
@@ -223,9 +222,8 @@ namespace MQTTnet.Client
 
             ThrowIfDisposed();
             ThrowIfNotConnected();
-
+            
-            var unsubscribeFactory = new MqttUnsubscribePacketFactory();
+            var unsubscribePacket = _packetFactories.Unsubscribe.Create(options);
-            var unsubscribePacket = unsubscribeFactory.Create(options);
             unsubscribePacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
 
             var unsubAckPacket = await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, cancellationToken).ConfigureAwait(false);
@@ -243,10 +241,7 @@ namespace MQTTnet.Client
             ThrowIfDisposed();
             ThrowIfNotConnected();
 
-            var publishPacketFactory = new MqttPublishPacketFactory();
+            var publishPacket = _packetFactories.Publish.Create(applicationMessage);
-            var publishPacket = publishPacketFactory.Create(applicationMessage);
-            
-            //_adapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage);
 
             switch (applicationMessage.QualityOfServiceLevel)
             {
@@ -303,8 +298,7 @@ namespace MQTTnet.Client
 
             try
             {
-                var connectPacketFactory = new MqttConnectPacketFactory();
+                var connectPacket = _packetFactories.Connect.Create(options);
-                var connectPacket = connectPacketFactory.Create(options);
 
                 var connAckPacket = await SendAndReceiveAsync<MqttConnAckPacket>(connectPacket, cancellationToken).ConfigureAwait(false);
 
@@ -336,7 +330,10 @@ namespace MQTTnet.Client
 
         void ThrowIfConnected(string message)
         {
-            if (IsConnected) throw new MqttProtocolViolationException(message);
+            if (IsConnected)
+            {
+                throw new MqttProtocolViolationException(message);
+            }
         }
 
         Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientConnectResult connectResult)
@@ -360,7 +357,11 @@ namespace MQTTnet.Client
                 if (_adapter != null)
                 {
                     _logger.Verbose("Disconnecting [Timeout={0}]", Options.CommunicationTimeout);
-                    await _adapter.DisconnectAsync(Options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
+
+                    using (var timeout = new CancellationTokenSource(Options.CommunicationTimeout))
+                    {
+                        await _adapter.DisconnectAsync(timeout.Token).ConfigureAwait(false);
+                    }
                 }
 
                 _logger.Verbose("Disconnected from adapter.");
@@ -434,7 +435,7 @@ namespace MQTTnet.Client
                 packetIdentifier = packetWithIdentifier.PacketIdentifier;
             }
 
-            using (var packetAwaiter = _packetDispatcher.AddAwaitable<TResponsePacket>(packetIdentifier))
+            using (var packetAwaitable = _packetDispatcher.AddAwaitable<TResponsePacket>(packetIdentifier))
             {
                 try
                 {
@@ -443,12 +444,12 @@ namespace MQTTnet.Client
                 catch (Exception exception)
                 {
                     _logger.Warning(exception, "Error when sending request packet ({0}).", requestPacket.GetType().Name);
-                    packetAwaiter.Fail(exception);
+                    packetAwaitable.Fail(exception);
                 }
 
                 try
                 {
-                    return await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false);
+                    return await packetAwaitable.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false);
                 }
                 catch (Exception exception)
                 {
@@ -686,8 +687,7 @@ namespace MQTTnet.Client
             {
                 if (!eventArgs.ProcessingFailed)
                 {
-                    var pubAckPacketFactory = new MqttPubAckPacketFactory();
+                    var pubAckPacket = _packetFactories.PubAck.Create(eventArgs);
-                    var pubAckPacket = pubAckPacketFactory.Create(eventArgs);
                     return SendAsync(pubAckPacket, cancellationToken);
                 }
             }
@@ -695,8 +695,7 @@ namespace MQTTnet.Client
             {
                 if (!eventArgs.ProcessingFailed)
                 {
-                    var pubRecPacketFactory = new MqttPubRecPacketFactory();
+                    var pubRecPacket = _packetFactories.PubRec.Create(eventArgs);
-                    var pubRecPacket = pubRecPacketFactory.Create(eventArgs);
                     return SendAsync(pubRecPacket, cancellationToken);
                 }
             }
@@ -714,8 +713,7 @@ namespace MQTTnet.Client
             {
                 // The packet is unknown. Probably due to a restart of the client.
                 // So wen send this to the server to trigger a full resend of the message.
-                var pubRelPacketFactory = new MqttPubRelPacketFactory();
+                var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
-                var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.PacketIdentifierNotFound);
                 return SendAsync(pubRelPacket, cancellationToken);
             }
 
@@ -724,8 +722,7 @@ namespace MQTTnet.Client
 
         Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
         {
-            var pubCompPacketFactory = new MqttPubCompPacketFactory();
+            var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            var pubCompPacket =pubCompPacketFactory.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
             return SendAsync(pubCompPacket, cancellationToken);
         }
 
@@ -774,8 +771,7 @@ namespace MQTTnet.Client
 
             var pubRecPacket = await SendAndReceiveAsync<MqttPubRecPacket>(publishPacket, cancellationToken).ConfigureAwait(false);
 
-            var pubRelPacketFactory = new MqttPubRelPacketFactory();
+            var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
 
             var pubCompPacket = await SendAndReceiveAsync<MqttPubCompPacket>(pubRelPacket, cancellationToken).ConfigureAwait(false);
 

Source/MQTTnet/Client/Publishing/MqttClientPublishResult.cs

@@ -1,12 +1,14 @@
-
+using System.Collections.Generic;
-using System.Collections.Generic;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Client.Publishing
 {
     public sealed class MqttClientPublishResult
     {
-        public ushort? PacketIdentifier { get; set; }
+        /// <summary>
+        /// Gets the packet identifier which was used for this publish.
+        /// </summary>
+        public ushort PacketIdentifier { get; set; }
 
         /// <summary>
         /// Gets or sets the reason code.

Source/MQTTnet/Formatter/MqttConnectPacketFactory.cs

@@ -17,7 +17,6 @@ namespace MQTTnet.Formatter
                 Password = clientOptions.Credentials?.Password,
                 CleanSession = clientOptions.CleanSession,
                 KeepAlivePeriod = (ushort) clientOptions.KeepAlivePeriod.TotalSeconds,
-                WillMessage = clientOptions.WillMessage,
                 Properties =
                 {
                     AuthenticationMethod = clientOptions.AuthenticationMethod,
@@ -32,6 +31,26 @@ namespace MQTTnet.Formatter
                 }
             };
 
+            if (clientOptions.WillMessage != null)
+            {
+                connectPacket.WillFlag = true;
+                connectPacket.WillTopic = clientOptions.WillMessage.Topic;
+                connectPacket.WillQoS = clientOptions.WillMessage.QualityOfServiceLevel;
+                connectPacket.WillMessage = clientOptions.WillMessage.Payload;
+                connectPacket.WillRetain = clientOptions.WillMessage.Retain;
+                connectPacket.WillProperties.ContentType = clientOptions.WillMessage.ContentType;
+                connectPacket.WillProperties.CorrelationData = clientOptions.WillMessage.CorrelationData;
+                connectPacket.WillProperties.ResponseTopic = clientOptions.WillMessage.ResponseTopic;
+                connectPacket.WillProperties.MessageExpiryInterval = clientOptions.WillMessage.MessageExpiryInterval;
+                connectPacket.WillProperties.PayloadFormatIndicator = clientOptions.WillMessage.PayloadFormatIndicator;
+                //connectPacket.WillProperties.WillDelayInterval = clientOptions.WillMessage.;
+
+                if (clientOptions.WillMessage.UserProperties != null)
+                {
+                    connectPacket.WillProperties.UserProperties.AddRange(clientOptions.WillMessage.UserProperties);
+                }
+            }
+
             if (clientOptions.UserProperties != null)
             {
                 connectPacket.Properties.UserProperties.AddRange(clientOptions.UserProperties);

Source/MQTTnet/Formatter/MqttPacketFactories.cs

@@ -0,0 +1,29 @@
+namespace MQTTnet.Formatter
+{
+    public sealed class MqttPacketFactories
+    {
+        public MqttConnectPacketFactory Connect { get; } = new MqttConnectPacketFactory();
+        
+        public MqttConnAckPacketFactory ConnAck { get; } = new MqttConnAckPacketFactory();
+
+        public MqttDisconnectPacketFactory Disconnect { get; } = new MqttDisconnectPacketFactory();
+
+        public MqttSubscribePacketFactory Subscribe { get; } = new MqttSubscribePacketFactory();
+
+        public MqttSubAckPacketFactory SubAck { get; } = new MqttSubAckPacketFactory();
+
+        public MqttUnsubscribePacketFactory Unsubscribe { get; } = new MqttUnsubscribePacketFactory();
+
+        public MqttUnsubAckPacketFactory UnsubAck { get; } = new MqttUnsubAckPacketFactory();
+
+        public MqttPublishPacketFactory Publish { get; } = new MqttPublishPacketFactory();
+
+        public MqttPubAckPacketFactory PubAck { get; } = new MqttPubAckPacketFactory();
+
+        public MqttPubRelPacketFactory PubRel { get; } = new MqttPubRelPacketFactory();
+        
+        public MqttPubRecPacketFactory PubRec { get; } = new MqttPubRecPacketFactory();
+        
+        public MqttPubCompPacketFactory PubComp { get; } = new MqttPubCompPacketFactory();
+    }
+}

Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs

@@ -1,4 +1,5 @@
 using System;
+using MQTTnet.Exceptions;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Formatter
@@ -41,5 +42,36 @@ namespace MQTTnet.Formatter
 
             return packet;
         }
+
+        public MqttPublishPacket Create(MqttConnectPacket connectPacket)
+        {
+            if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket));
+
+            if (!connectPacket.WillFlag)
+            {
+                throw new MqttProtocolViolationException("The CONNECT packet contains no will message (WillFlag).");
+            }
+            
+            var packet = new MqttPublishPacket
+            {
+                Topic = connectPacket.WillTopic,
+                Payload = connectPacket.WillMessage,
+                QualityOfServiceLevel = connectPacket.WillQoS,
+                Retain = connectPacket.WillRetain,
+                Dup = false,
+                Properties =
+                {
+                    ContentType = connectPacket.WillProperties.ContentType,
+                    CorrelationData = connectPacket.WillProperties.CorrelationData,
+                    MessageExpiryInterval = connectPacket.WillProperties.MessageExpiryInterval,
+                    PayloadFormatIndicator = connectPacket.WillProperties.PayloadFormatIndicator,
+                    ResponseTopic = connectPacket.WillProperties.ResponseTopic
+                }
+            };
+            
+            packet.Properties.UserProperties.AddRange(connectPacket.WillProperties.UserProperties);
+
+            return packet;
+        }
     }
 }

Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs

@@ -11,7 +11,7 @@ namespace MQTTnet.Formatter.V3
     public class MqttV310PacketFormatter : IMqttPacketFormatter
     {
         const int FixedHeaderSize = 1;
-        
+
         static readonly MqttDisconnectPacket DisconnectPacket = new MqttDisconnectPacket();
 
         readonly IMqttPacketWriter _packetWriter;
@@ -30,7 +30,7 @@ namespace MQTTnet.Formatter.V3
             _packetWriter.Seek(5);
 
             var fixedHeader = EncodePacket(packet, _packetWriter);
-            var remainingLength = (uint)(_packetWriter.Length - 5);
+            var remainingLength = (uint) (_packetWriter.Length - 5);
 
             var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
 
@@ -57,7 +57,7 @@ namespace MQTTnet.Formatter.V3
                 throw new MqttProtocolViolationException($"The packet type is invalid ({controlPacketType}).");
             }
 
-            switch ((MqttControlPacketType)controlPacketType)
+            switch ((MqttControlPacketType) controlPacketType)
             {
                 case MqttControlPacketType.Connect: return DecodeConnectPacket(receivedMqttPacket.BodyReader);
                 case MqttControlPacketType.ConnAck: return DecodeConnAckPacket(receivedMqttPacket.BodyReader);
@@ -187,7 +187,7 @@ namespace MQTTnet.Formatter.V3
                 var topicFilter = new MqttTopicFilter
                 {
                     Topic = body.ReadStringWithLengthPrefix(),
-                    QualityOfServiceLevel = (MqttQualityOfServiceLevel)body.ReadByte()
+                    QualityOfServiceLevel = (MqttQualityOfServiceLevel) body.ReadByte()
                 };
 
                 packet.TopicFilters.Add(topicFilter);
@@ -201,7 +201,7 @@ namespace MQTTnet.Formatter.V3
             ThrowIfBodyIsEmpty(receivedMqttPacket.BodyReader);
 
             var retain = (receivedMqttPacket.FixedHeader & 0x1) > 0;
-            var qualityOfServiceLevel = (MqttQualityOfServiceLevel)(receivedMqttPacket.FixedHeader >> 1 & 0x3);
+            var qualityOfServiceLevel = (MqttQualityOfServiceLevel) (receivedMqttPacket.FixedHeader >> 1 & 0x3);
             var dup = (receivedMqttPacket.FixedHeader & 0x8) > 0;
 
             var topic = receivedMqttPacket.BodyReader.ReadStringWithLengthPrefix();
@@ -267,13 +267,12 @@ namespace MQTTnet.Formatter.V3
 
             if (willFlag)
             {
-                packet.WillMessage = new MqttApplicationMessage
+                packet.WillFlag = true;
-                {
+                packet.WillQoS = (MqttQualityOfServiceLevel) willQoS;
-                    Topic = body.ReadStringWithLengthPrefix(),
+                packet.WillRetain = willRetain;
-                    Payload = body.ReadWithLengthPrefix(),
+
-                    QualityOfServiceLevel = (MqttQualityOfServiceLevel)willQoS,
+                packet.WillTopic = body.ReadStringWithLengthPrefix();
-                    Retain = willRetain
+                packet.WillMessage = body.ReadWithLengthPrefix();
-                };
             }
 
             if (usernameFlag)
@@ -301,7 +300,7 @@ namespace MQTTnet.Formatter.V3
 
             while (!body.EndOfStream)
             {
-                packet.ReturnCodes.Add((MqttSubscribeReturnCode)body.ReadByte());
+                packet.ReturnCodes.Add((MqttSubscribeReturnCode) body.ReadByte());
             }
 
             return packet;
@@ -314,7 +313,7 @@ namespace MQTTnet.Formatter.V3
             var packet = new MqttConnAckPacket();
 
             body.ReadByte(); // Reserved.
-            packet.ReturnCode = (MqttConnectReturnCode)body.ReadByte();
+            packet.ReturnCode = (MqttConnectReturnCode) body.ReadByte();
 
             return packet;
         }
@@ -351,12 +350,12 @@ namespace MQTTnet.Formatter.V3
                 connectFlags |= 0x2;
             }
 
-            if (packet.WillMessage != null)
+            if (packet.WillFlag)
             {
                 connectFlags |= 0x4;
-                connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3);
+                connectFlags |= (byte) ((byte) packet.WillQoS << 3);
 
-                if (packet.WillMessage.Retain)
+                if (packet.WillRetain)
                 {
                     connectFlags |= 0x20;
                 }
@@ -381,10 +380,10 @@ namespace MQTTnet.Formatter.V3
             packetWriter.Write(packet.KeepAlivePeriod);
             packetWriter.WriteWithLengthPrefix(packet.ClientId);
 
-            if (packet.WillMessage != null)
+            if (packet.WillFlag)
             {
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic);
+                packetWriter.WriteWithLengthPrefix(packet.WillTopic);
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload);
+                packetWriter.WriteWithLengthPrefix(packet.WillMessage);
             }
 
             if (packet.Username != null)
@@ -403,7 +402,7 @@ namespace MQTTnet.Formatter.V3
         protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter)
         {
             packetWriter.Write(0); // Reserved.
-            packetWriter.Write((byte)packet.ReturnCode);
+            packetWriter.Write((byte) packet.ReturnCode);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
         }
@@ -455,7 +454,7 @@ namespace MQTTnet.Formatter.V3
                 fixedHeader |= 0x01;
             }
 
-            fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1);
+            fixedHeader |= (byte) ((byte) packet.QualityOfServiceLevel << 1);
 
             if (packet.Dup)
             {
@@ -525,14 +524,14 @@ namespace MQTTnet.Formatter.V3
                     {
                         throw new MqttProtocolViolationException("RetainAsPublished is not supported in 3.1.1.");
                     }
-                    
+
                     if (topicFilter.RetainHandling != MqttRetainHandling.SendAtSubscribe)
                     {
                         throw new MqttProtocolViolationException("RetainHandling is not supported in 3.1.1.");
                     }
-                    
+
                     packetWriter.WriteWithLengthPrefix(topicFilter.Topic);
-                    packetWriter.Write((byte)topicFilter.QualityOfServiceLevel);
+                    packetWriter.Write((byte) topicFilter.QualityOfServiceLevel);
                 }
             }
 
@@ -552,7 +551,7 @@ namespace MQTTnet.Formatter.V3
             {
                 foreach (var packetSubscribeReturnCode in packet.ReturnCodes)
                 {
-                    packetWriter.Write((byte)packetSubscribeReturnCode);
+                    packetWriter.Write((byte) packetSubscribeReturnCode);
                 }
             }
 
@@ -606,4 +605,4 @@ namespace MQTTnet.Formatter.V3
             }
         }
     }
-}
+}

Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs

@@ -24,12 +24,12 @@ namespace MQTTnet.Formatter.V3
                 connectFlags |= 0x2;
             }
 
-            if (packet.WillMessage != null)
+            if (packet.WillFlag)
             {
                 connectFlags |= 0x4;
-                connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3);
+                connectFlags |= (byte)((byte)packet.WillQoS << 3);
 
-                if (packet.WillMessage.Retain)
+                if (packet.WillRetain)
                 {
                     connectFlags |= 0x20;
                 }
@@ -54,10 +54,10 @@ namespace MQTTnet.Formatter.V3
             packetWriter.Write(packet.KeepAlivePeriod);
             packetWriter.WriteWithLengthPrefix(packet.ClientId);
 
-            if (packet.WillMessage != null)
+            if (packet.WillFlag)
             {
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic);
+                packetWriter.WriteWithLengthPrefix(packet.WillTopic);
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload);
+                packetWriter.WriteWithLengthPrefix(packet.WillMessage);
             }
 
             if (packet.Username != null)

Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Generic;
 using MQTTnet.Adapter;
 using MQTTnet.Exceptions;
 using MQTTnet.Packets;
@@ -72,11 +71,9 @@ namespace MQTTnet.Formatter.V5
 
             if (willMessageFlag)
             {
-                packet.WillMessage = new MqttApplicationMessage
+                packet.WillFlag = true;
-                {
+                packet.WillQoS = (MqttQualityOfServiceLevel) willMessageQoS;
-                    QualityOfServiceLevel = (MqttQualityOfServiceLevel)willMessageQoS,
+                packet.WillRetain = willMessageRetainFlag;
-                    Retain = willMessageRetainFlag
-                };
             }
 
             packet.KeepAlivePeriod = body.ReadTwoByteInteger();
@@ -128,7 +125,7 @@ namespace MQTTnet.Formatter.V5
 
             packet.ClientId = body.ReadStringWithLengthPrefix();
 
-            if (packet.WillMessage != null)
+            if (packet.WillFlag)
             {
                 var willPropertiesReader = new MqttV500PropertiesReader(body);
 
@@ -136,50 +133,31 @@ namespace MQTTnet.Formatter.V5
                 {
                     if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.PayloadFormatIndicator)
                     {
-                        packet.WillMessage.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator();
+                        packet.WillProperties.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator();
                     }
                     else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.MessageExpiryInterval)
                     {
-                        packet.WillMessage.MessageExpiryInterval = propertiesReader.ReadMessageExpiryInterval();
+                        packet.WillProperties.MessageExpiryInterval = propertiesReader.ReadMessageExpiryInterval();
-                    }
-                    else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.TopicAlias)
-                    {
-                        packet.WillMessage.TopicAlias = propertiesReader.ReadTopicAlias();
                     }
                     else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ResponseTopic)
                     {
-                        packet.WillMessage.ResponseTopic = propertiesReader.ReadResponseTopic();
+                        packet.WillProperties.ResponseTopic = propertiesReader.ReadResponseTopic();
                     }
                     else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.CorrelationData)
                     {
-                        packet.WillMessage.CorrelationData = propertiesReader.ReadCorrelationData();
+                        packet.WillProperties.CorrelationData = propertiesReader.ReadCorrelationData();
-                    }
-                    else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier)
-                    {
-                        if (packet.WillMessage.SubscriptionIdentifiers == null)
-                        {
-                            packet.WillMessage.SubscriptionIdentifiers = new List<uint>();
-                        }
-
-                        packet.WillMessage.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier());
                     }
                     else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.ContentType)
                     {
-                        packet.WillMessage.ContentType = propertiesReader.ReadContentType();
+                        packet.WillProperties.ContentType = propertiesReader.ReadContentType();
                     }
                     else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.WillDelayInterval)
                     {
-                        // This is a special case!
+                        packet.WillProperties.WillDelayInterval = propertiesReader.ReadWillDelayInterval();
-                        packet.Properties.WillDelayInterval = propertiesReader.ReadWillDelayInterval();
                     }
                     else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                     {
-                        if (packet.WillMessage.UserProperties == null)
+                        propertiesReader.AddUserPropertyTo(packet.WillProperties.UserProperties);
-                        {
-                            packet.WillMessage.UserProperties = new List<MqttUserProperty>();
-                        }
-
-                        propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                     }
                     else
                     {
@@ -187,8 +165,8 @@ namespace MQTTnet.Formatter.V5
                     }
                 }
 
-                packet.WillMessage.Topic = body.ReadStringWithLengthPrefix();
+                packet.WillTopic = body.ReadStringWithLengthPrefix();
-                packet.WillMessage.Payload = body.ReadWithLengthPrefix();
+                packet.WillMessage = body.ReadWithLengthPrefix();
             }
 
             if (usernameFlag)

Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs

@@ -88,12 +88,12 @@ namespace MQTTnet.Formatter.V5
                 connectFlags |= 0x2;
             }
 
-            if (packet.WillMessage != null)
+            if (packet.WillFlag)
             {
                 connectFlags |= 0x4;
-                connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3);
+                connectFlags |= (byte)((byte)packet.WillQoS << 3);
 
-                if (packet.WillMessage.Retain)
+                if (packet.WillRetain)
                 {
                     connectFlags |= 0x20;
                 }
@@ -135,25 +135,23 @@ namespace MQTTnet.Formatter.V5
 
             packetWriter.WriteWithLengthPrefix(packet.ClientId);
             
-            if (packet.WillMessage != null)
+            if (packet.WillFlag)
             {
                 var willPropertiesWriter = new MqttV500PropertiesWriter();
-                willPropertiesWriter.WritePayloadFormatIndicator(packet.WillMessage.PayloadFormatIndicator);
+                willPropertiesWriter.WritePayloadFormatIndicator(packet.WillProperties.PayloadFormatIndicator);
-                willPropertiesWriter.WriteMessageExpiryInterval(packet.WillMessage.MessageExpiryInterval);
+                willPropertiesWriter.WriteMessageExpiryInterval(packet.WillProperties.MessageExpiryInterval);
-                willPropertiesWriter.WriteTopicAlias(packet.WillMessage.TopicAlias);
+                willPropertiesWriter.WriteResponseTopic(packet.WillProperties.ResponseTopic);
-                willPropertiesWriter.WriteResponseTopic(packet.WillMessage.ResponseTopic);
+                willPropertiesWriter.WriteCorrelationData(packet.WillProperties.CorrelationData);
-                willPropertiesWriter.WriteCorrelationData(packet.WillMessage.CorrelationData);
+                willPropertiesWriter.WriteContentType(packet.WillProperties.ContentType);
-                willPropertiesWriter.WriteSubscriptionIdentifiers(packet.WillMessage.SubscriptionIdentifiers);
+                willPropertiesWriter.WriteUserProperties(packet.WillProperties.UserProperties);
-                willPropertiesWriter.WriteContentType(packet.WillMessage.ContentType);
-                willPropertiesWriter.WriteUserProperties(packet.WillMessage.UserProperties);
                 
                 // This is a special case!
                 willPropertiesWriter.WriteWillDelayInterval(packet.Properties?.WillDelayInterval);
 
                 willPropertiesWriter.WriteTo(packetWriter);
 
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic);
+                packetWriter.WriteWithLengthPrefix(packet.WillTopic);
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload);
+                packetWriter.WriteWithLengthPrefix(packet.WillMessage);
             }
 
             if (packet.Username != null)

Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs

@@ -143,7 +143,7 @@ namespace MQTTnet.Formatter.V5
             return _body.ReadVariableLengthInteger();
         }
 
-        public MqttPayloadFormatIndicator? ReadPayloadFormatIndicator()
+        public MqttPayloadFormatIndicator ReadPayloadFormatIndicator()
         {
             return (MqttPayloadFormatIndicator)_body.ReadByte();
         }

Source/MQTTnet/Internal/MqttPacketBus.cs

@@ -11,11 +11,11 @@ namespace MQTTnet.Internal
     {
         readonly object _syncRoot = new object();
 
-        readonly Queue<MqttPacketBusItem>[] _partitions = new Queue<MqttPacketBusItem>[]
+        readonly LinkedList<MqttPacketBusItem>[] _partitions = new LinkedList<MqttPacketBusItem>[]
         {
-            new Queue<MqttPacketBusItem>(4096),
+            new LinkedList<MqttPacketBusItem>(),
-            new Queue<MqttPacketBusItem>(1028),
+            new LinkedList<MqttPacketBusItem>(),
-            new Queue<MqttPacketBusItem>(128)
+            new LinkedList<MqttPacketBusItem>()
         };
 
         readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
@@ -33,11 +33,14 @@ namespace MQTTnet.Internal
             }
         }
 
-        public int DataPacketsCount()
+        public int DataPacketsCount
         {
-            lock (_syncRoot)
+            get
             {
-                return _partitions[(int)MqttPacketBusPartition.Data].Count;
+                lock (_syncRoot)
+                {
+                    return _partitions[(int) MqttPacketBusPartition.Data].Count;
+                }
             }
         }
 
@@ -55,7 +58,7 @@ namespace MQTTnet.Internal
 
             lock (_syncRoot)
             {
-                _partitions[(int) partition].Enqueue(item);
+                _partitions[(int) partition].AddLast(item);
             }
 
             _semaphore.Release();
@@ -85,7 +88,9 @@ namespace MQTTnet.Internal
 
                         if (_partitions[_activePartition].Count > 0)
                         {
-                            return _partitions[_activePartition].Dequeue();
+                            var item = _partitions[_activePartition].First;
+                            _partitions[_activePartition].RemoveFirst();
+                            return item.Value;
                         }
                     }
                 }
@@ -102,5 +107,16 @@ namespace MQTTnet.Internal
         {
             _semaphore?.Dispose();
         }
+
+        public void Clear()
+        {
+            lock (_syncRoot)
+            {
+                foreach (var partition in _partitions)
+                {
+                    partition.Clear();
+                }
+            }
+        }
     }
 }

Source/MQTTnet/Internal/MqttTaskTimeout.cs

@@ -30,29 +30,5 @@ namespace MQTTnet.Internal
                 }
             }
         }
-
-        public static async Task<TResult> WaitAsync<TResult>(Func<CancellationToken, Task<TResult>> action, TimeSpan timeout, CancellationToken cancellationToken)
-        {
-            if (action == null) throw new ArgumentNullException(nameof(action));
-
-            using (var timeoutCts = new CancellationTokenSource(timeout))
-            using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken))
-            {
-                try
-                {
-                    return await action(linkedCts.Token).ConfigureAwait(false);
-                }
-                catch (OperationCanceledException exception)
-                {
-                    var timeoutReached = timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested;
-                    if (timeoutReached)
-                    {
-                        throw new MqttCommunicationTimedOutException(exception);
-                    }
-
-                    throw;
-                }
-            }
-        }     
     }
 }

Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs

@@ -113,7 +113,7 @@ namespace MQTTnet.LowLevelClient
         {
             try
             {
-                await _adapter.DisconnectAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+                await _adapter.DisconnectAsync(cancellationToken).ConfigureAwait(false);
             }
             catch (Exception exception)
             {

Source/MQTTnet/MqttApplicationMessage.cs

@@ -51,7 +51,7 @@ namespace MQTTnet
         /// The feature is very similar to the HTTP header concept.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        public List<MqttUserProperty> UserProperties { get; set; }
+        public List<MqttUserProperty> UserProperties { get; set; } = new List<MqttUserProperty>();
 
         /// <summary>
         /// Gets or sets the content type.
@@ -74,7 +74,7 @@ namespace MQTTnet
         /// If no payload format indicator is provided, the default value is 0.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        public MqttPayloadFormatIndicator? PayloadFormatIndicator { get; set; }
+        public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified;
 
         /// <summary>
         /// Gets or sets the message expiry interval.
@@ -84,7 +84,7 @@ namespace MQTTnet
         /// When the retained=true option is set on the PUBLISH message, this interval also defines how long a message is retained on a topic.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        public uint? MessageExpiryInterval { get; set; }
+        public uint MessageExpiryInterval { get; set; }
 
         /// <summary>
         /// Gets or sets the topic alias.

Source/MQTTnet/MqttApplicationMessageBuilder.cs

@@ -87,7 +87,7 @@ namespace MQTTnet
         /// When the retained=true option is set on the PUBLISH message, this interval also defines how long a message is retained on a topic.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        uint? _messageExpiryInterval;
+        uint _messageExpiryInterval;
 
         /// <summary>
         /// The payload format indicator.
@@ -97,7 +97,7 @@ namespace MQTTnet
         /// If no payload format indicator is provided, the default value is 0.
         /// Hint: MQTT 5 feature only.
         /// </summary>
-        MqttPayloadFormatIndicator? _payloadFormatIndicator;
+        MqttPayloadFormatIndicator _payloadFormatIndicator;
 
         /// <summary>
         /// The user properties.

Source/MQTTnet/Packets/MqttConnectPacket.cs

@@ -1,4 +1,6 @@
-namespace MQTTnet.Packets
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Packets
 {
     public sealed class MqttConnectPacket : MqttBasePacket
     {
@@ -14,13 +16,23 @@
        /// Also called "Clean Start" in MQTTv5.
        /// </summary>
         public bool CleanSession { get; set; }
-
-        public MqttApplicationMessage WillMessage { get; set; }
         
         /// <summary>
         /// Added in MQTTv5.
         /// </summary>
         public MqttConnectPacketProperties Properties { get; } = new MqttConnectPacketProperties();
+
+        public bool WillFlag { get; set; }
+        
+        public string WillTopic { get; set; }
+        
+        public byte[] WillMessage { get; set; }
+        
+        public MqttQualityOfServiceLevel WillQoS { get; set; } = MqttQualityOfServiceLevel.AtMostOnce;
+        
+        public bool WillRetain { get; set; }
+
+        public MqttConnectPacketWillProperties WillProperties { get; } = new MqttConnectPacketWillProperties();
         
         public override string ToString()
         {

Source/MQTTnet/Packets/MqttConnectPacketWillProperties.cs

@@ -0,0 +1,22 @@
+using System.Collections.Generic;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Packets
+{
+    public sealed class MqttConnectPacketWillProperties
+    {
+        public uint WillDelayInterval { get; set; }
+
+        public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified;
+        
+        public uint MessageExpiryInterval { get; set; }
+        
+        public string ContentType { get; set; }
+        
+        public string ResponseTopic { get; set; }
+        
+        public byte[] CorrelationData { get; set; }
+
+        public List<MqttUserProperty> UserProperties { get; } = new List<MqttUserProperty>();
+    }
+}

Source/MQTTnet/Packets/MqttPublishPacketProperties.cs

@@ -5,9 +5,9 @@ namespace MQTTnet.Packets
 {
     public sealed class MqttPublishPacketProperties
     {
-        public MqttPayloadFormatIndicator? PayloadFormatIndicator { get; set; }
+        public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; } = MqttPayloadFormatIndicator.Unspecified;
 
-        public uint? MessageExpiryInterval { get; set; }
+        public uint MessageExpiryInterval { get; set; }
 
         public ushort? TopicAlias { get; set; }
 

Source/MQTTnet/Server/Internal/MqttClientConnection.cs

@@ -19,12 +19,13 @@ namespace MQTTnet.Server.Internal
     {
         readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
         readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
+        readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
         
         readonly MqttClientSessionsManager _sessionsManager;
-        
+
         readonly MqttNetSourceLogger _logger;
         readonly IMqttServerOptions _serverOptions;
-        
+
         readonly MqttConnectPacket _connectPacket;
 
         CancellationTokenSource _cancellationToken;
@@ -39,10 +40,10 @@ namespace MQTTnet.Server.Internal
         {
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
-            
+
             ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
             Endpoint = channelAdapter.Endpoint;
-            
+
             Session = session ?? throw new ArgumentNullException(nameof(session));
             _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
 
@@ -50,10 +51,10 @@ namespace MQTTnet.Server.Internal
             _logger = logger.WithSource(nameof(MqttClientConnection));
         }
 
-        public string ClientId => _connectPacket.ClientId;
+        public string Id => _connectPacket.ClientId;
+
+        public string Endpoint { get; }
 
-        public string Endpoint {get;}
-        
         public IMqttChannelAdapter ChannelAdapter { get; }
 
         public MqttClientConnectionStatistics Statistics { get; } = new MqttClientConnectionStatistics();
@@ -61,13 +62,13 @@ namespace MQTTnet.Server.Internal
         public bool IsRunning { get; private set; }
 
         public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod;
-        
+
         public MqttClientSession Session { get; }
 
         public bool IsTakenOver { get; set; }
 
         public bool IsCleanDisconnect { get; private set; }
-        
+
         public async Task StopAsync(MqttClientDisconnectReason reason)
         {
             IsRunning = false;
@@ -88,7 +89,7 @@ namespace MQTTnet.Server.Internal
         {
             ChannelAdapter.ResetStatistics();
         }
-        
+
         public void Dispose()
         {
             _cancellationToken.Dispose();
@@ -96,10 +97,11 @@ namespace MQTTnet.Server.Internal
 
         public async Task RunAsync()
         {
-            _logger.Info("Client '{0}': Session started.", ClientId);
+            _logger.Info("Client '{0}': Session started.", Id);
+
+            Session.LatestConnectPacket = _connectPacket;
+            Session.WillMessageSent = false;
 
-            Session.WillMessage = _connectPacket.WillMessage;
-            
             using (var cancellationToken = new CancellationTokenSource())
             {
                 _cancellationToken = cancellationToken;
@@ -108,7 +110,8 @@ namespace MQTTnet.Server.Internal
                 {
                     Task.Run(() => SendPacketsLoop(cancellationToken.Token), cancellationToken.Token).RunInBackground(_logger);
 
-                    Session.IsCleanSession = false;
+                    // TODO: Check!.
+                    //Session.IsCleanSession = false;
 
                     IsRunning = true;
 
@@ -117,21 +120,22 @@ namespace MQTTnet.Server.Internal
                 finally
                 {
                     IsRunning = false;
-                    
+
                     cancellationToken.Cancel();
                     _cancellationToken = null;
                 }
             }
-        
+
             _packetDispatcher.CancelAll();
 
-            if (!IsTakenOver && !IsCleanDisconnect && Session.WillMessage != null)
+            if (!IsTakenOver && !IsCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent)
             {
-                _sessionsManager.DispatchApplicationMessage(Session.WillMessage, this);
+                var publishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket);
-                Session.WillMessage = null;
+                _sessionsManager.DispatchPublishPacket(publishPacket, this);
+                Session.WillMessageSent = true;
             }
 
-            _logger.Info("Client '{0}': Connection stopped.", ClientId);
+            _logger.Info("Client '{0}': Connection stopped.", Id);
         }
 
         Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
@@ -163,23 +167,19 @@ namespace MQTTnet.Server.Internal
                     }
                     else if (packet is MqttPubAckPacket pubAckPacket)
                     {
-                        // TODO: Do anything!
+                        Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier);
                     }
                     else if (packet is MqttPubCompPacket pubCompPacket)
                     {
-                        // TODO: Do anything!
+                        Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier);
                     }
                     else if (packet is MqttPubRecPacket pubRecPacket)
                     {
-                        // TODO: Do anything!
+                        HandleIncomingPubRecPacket(pubRecPacket);
-                        // TODO: Send pub rel!
-                        var pubRelPacketFactory = new MqttPubRelPacketFactory();
-                        var pubRelPacket = pubRelPacketFactory.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-                        Session.EnqueuePacket(new MqttPacketBusItem(pubRelPacket));
                     }
                     else if (packet is MqttPubRelPacket pubRelPacket)
                     {
-                        await HandleIncomingPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false);
+                        HandleIncomingPubRelPacket(pubRelPacket);
                     }
                     else if (packet is MqttSubscribePacket subscribePacket)
                     {
@@ -220,30 +220,25 @@ namespace MQTTnet.Server.Internal
             {
                 if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", ClientId);
+                    _logger.Warning(exception, "Client '{0}': Communication exception while receiving client packets.", Id);
                 }
                 else
                 {
-                    _logger.Error(exception, "Client '{0}': Error while receiving client packets.", ClientId);
+                    _logger.Error(exception, "Client '{0}': Error while receiving client packets.", Id);
                 }
             }
         }
 
         async Task SendPacketsLoop(CancellationToken cancellationToken)
         {
-            //MqttQueuedApplicationMessage queuedApplicationMessage = null;
-
             MqttPacketBusItem packetBusItem = null;
-            //MqttPublishPacket publishPacket = null;
 
             try
             {
                 while (!cancellationToken.IsCancellationRequested)
                 {
-                    //queuedApplicationMessage = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false);
-                    
                     packetBusItem = await Session.DequeuePacketAsync(cancellationToken).ConfigureAwait(false);
-                    
+
                     // Also check the cancellation token here because the dequeue is blocking and may take some time.
                     if (cancellationToken.IsCancellationRequested)
                     {
@@ -254,7 +249,7 @@ namespace MQTTnet.Server.Internal
                     {
                         await SendPacketAsync(packetBusItem.Packet, cancellationToken).ConfigureAwait(false);
                         packetBusItem.MarkAsDelivered();
-                        
+
                         _logger.Verbose("Sent packet '{0}'.", packetBusItem.Packet);
                     }
                     catch (OperationCanceledException)
@@ -265,7 +260,7 @@ namespace MQTTnet.Server.Internal
                     {
                         packetBusItem.MarkAsFailed(exception);
                     }
-                    
+
                     // TODO: Move QoS 1 + 2 to other layer!
                     // if (queuedApplicationMessage == null)
                     // {
@@ -305,7 +300,7 @@ namespace MQTTnet.Server.Internal
                     //     }
                     // }
 
-                    _logger.Verbose("Client '{0}': Queued application message sent.", ClientId);
+                    _logger.Verbose("Client '{0}': Queued application message sent.", Id);
                 }
             }
             catch (OperationCanceledException)
@@ -315,15 +310,15 @@ namespace MQTTnet.Server.Internal
             {
                 if (exception is MqttCommunicationTimedOutException)
                 {
-                    _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Timeout.", ClientId);
+                    _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Timeout.", Id);
                 }
                 else if (exception is MqttCommunicationException)
                 {
-                    _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Communication exception.", ClientId);
+                    _logger.Warning(exception, "Client '{0}': Sending publish packet failed: Communication exception.", Id);
                 }
                 else
                 {
-                    _logger.Error(exception, "Client '{0}': Sending publish packet failed.", ClientId);
+                    _logger.Error(exception, "Client '{0}': Sending publish packet failed.", Id);
                 }
 
                 if (packetBusItem?.Packet is MqttPublishPacket publishPacket)
@@ -333,23 +328,8 @@ namespace MQTTnet.Server.Internal
                         publishPacket.Dup = true;
                         Session.EnqueuePacket(new MqttPacketBusItem(publishPacket));
                     }
-                    
-                    // if (queuedApplicationMessage != null)
-                    // {
-                    //     queuedApplicationMessage.IsDuplicate = true;
-                    //     Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage);
-                    // }
                 }
                 
-                // if (publishPacket?.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
-                // {
-                //     if (queuedApplicationMessage != null)
-                //     {
-                //         queuedApplicationMessage.IsDuplicate = true;
-                //         Session.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage);
-                //     }
-                // }
-
                 StopInternal();
             }
         }
@@ -387,35 +367,35 @@ namespace MQTTnet.Server.Internal
             _cancellationToken?.Cancel();
         }
         
-        Task HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
+        void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
         {
-            var pubCompPacketFactory = new MqttPubCompPacketFactory();
+            var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
-            var pubCompPacket =pubCompPacketFactory.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
+            Session.EnqueuePacket(new MqttPacketBusItem(pubRelPacket));
+        }
+
+        void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket)
+        {
+            var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success);
             Session.EnqueuePacket(new MqttPacketBusItem(pubCompPacket));
-            
-            //return SendPacketAsync(pubCompPacket, cancellationToken);
-            return PlatformAbstractionLayer.CompletedTask;
         }
 
         async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
             var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false);
 
-            var subAckPacketFactory = new MqttSubAckPacketFactory();
+            var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult);
-            var subAckPacket = subAckPacketFactory.Create(subscribePacket, subscribeResult);
 
             Session.EnqueuePacket(new MqttPacketBusItem(subAckPacket));
-            
+
             if (subscribeResult.CloseConnection)
             {
                 StopInternal();
                 return;
             }
 
-            var publishPacketFactory = new MqttPublishPacketFactory();
             foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages)
             {
-                var publishPacket = publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage);
+                var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
                 Session.EnqueuePacket(new MqttPacketBusItem(publishPacket));
             }
         }
@@ -424,9 +404,8 @@ namespace MQTTnet.Server.Internal
         {
             var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false);
 
-            var unsubAckPacketFactory = new MqttUnsubAckPacketFactory();
+            var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult);
-            var unsubAckPacket = unsubAckPacketFactory.Create(unsubscribePacket, unsubscribeResult);
+
-            
             Session.EnqueuePacket(new MqttPacketBusItem(unsubAckPacket));
 
             if (unsubscribeResult.CloseConnection)
@@ -439,12 +418,8 @@ namespace MQTTnet.Server.Internal
         {
             HandleTopicAlias(publishPacket);
 
-            var applicationMessageFactory = new MqttApplicationMessageFactory();
+            _sessionsManager.DispatchPublishPacket(publishPacket, this);
-            var applicationMessage = applicationMessageFactory.Create(publishPacket);
             
-            _sessionsManager.DispatchApplicationMessage(applicationMessage, this);
-
-            // TODO: Check!
             switch (publishPacket.QualityOfServiceLevel)
             {
                 case MqttQualityOfServiceLevel.AtMostOnce:
@@ -454,15 +429,13 @@ namespace MQTTnet.Server.Internal
                 }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                 {
-                    var pubAckPacketFactory = new MqttPubAckPacketFactory();
+                    var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
-                    var pubAckPacket = pubAckPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
                     Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket));
                     break;
                 }
                 case MqttQualityOfServiceLevel.ExactlyOnce:
                 {
-                    var pubRecPacketFactory = new MqttPubRecPacketFactory();
+                    var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
-                    var pubRecPacket = pubRecPacketFactory.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
                     Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket));
                     break;
                 }
@@ -500,7 +473,7 @@ namespace MQTTnet.Server.Internal
                 }
             }
         }
-        
+
         async Task TrySendDisconnectPacket(MqttClientDisconnectReason reason)
         {
             try
@@ -511,8 +484,7 @@ namespace MQTTnet.Server.Internal
                     ReasonString = reason.ToString()
                 };
 
-                var disconnectPacketFactory = new MqttDisconnectPacketFactory();
+                var disconnectPacket = _packetFactories.Disconnect.Create(disconnectOptions);
-                var disconnectPacket = disconnectPacketFactory.Create(disconnectOptions);
 
                 using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
                 {
@@ -521,10 +493,11 @@ namespace MQTTnet.Server.Internal
             }
             catch (Exception exception)
             {
-                _logger.Warning(exception, "Client '{{0}}': Error while sending DISCONNECT packet (Reason = {1}).", ClientId, reason);
+                _logger.Warning(exception, "Client '{{0}}': Error while sending DISCONNECT packet (Reason = {1}).", Id, reason);
             }
         }
 
+        // TODO: Fix!
         async Task<MqttPublishPacket> InvokeClientMessageQueueInterceptor(MqttPublishPacket publishPacket, MqttQueuedApplicationMessage queuedApplicationMessage)
         {
             if (_serverOptions.ClientMessageQueueInterceptor == null)
@@ -535,7 +508,7 @@ namespace MQTTnet.Server.Internal
             var context = new MqttClientMessageQueueInterceptorContext
             {
                 SenderClientId = queuedApplicationMessage.SenderClientId,
-                ReceiverClientId = ClientId,
+                ReceiverClientId = Id,
                 ApplicationMessage = queuedApplicationMessage.ApplicationMessage,
                 SubscriptionQualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel
             };

Source/MQTTnet/Server/Internal/MqttClientSession.cs

@@ -1,10 +1,12 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.Server.Internal
 {
@@ -12,6 +14,8 @@ namespace MQTTnet.Server.Internal
     {
         readonly MqttPacketBus _packetBus = new MqttPacketBus();
 
+        readonly Dictionary<ushort, MqttPublishPacket> _publishPacketsWithoutAcknowledge = new Dictionary<ushort, MqttPublishPacket>();
+
         readonly IMqttServerOptions _serverOptions;
         readonly MqttClientSessionsManager _clientSessionsManager;
 
@@ -24,52 +28,70 @@ namespace MQTTnet.Server.Internal
         {
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
-            ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            Id = clientId ?? throw new ArgumentNullException(nameof(clientId));
             Items = items ?? throw new ArgumentNullException(nameof(items));
 
             SubscriptionsManager = new MqttClientSubscriptionsManager(this, serverOptions, eventDispatcher, retainedMessagesManager);
-            //ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
         }
 
         public event EventHandler Deleted;
         
-        public string ClientId { get; }
+        public string Id { get; }
 
         public MqttPacketIdentifierProvider PacketIdentifierProvider { get; } = new MqttPacketIdentifierProvider();
 
         public DateTime CreatedTimestamp { get; } = DateTime.UtcNow;
         
-        public bool IsCleanSession { get; set; } = true;
+        //public bool IsCleanSession { get; set; } = true;
 
-        public MqttApplicationMessage WillMessage { get; set; }
+        public MqttConnectPacket LatestConnectPacket { get; set; }
 
         public MqttClientSubscriptionsManager SubscriptionsManager { get; }
         
-        //public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; }
-
         public IDictionary<object, object> Items { get; }
         
-        public void EnqueuePacket(MqttPacketBusItem packet)
+        public bool WillMessageSent { get; set; }
+
+        public long PendingDataPacketsCount => _packetBus.DataPacketsCount;
+
+        public void AcknowledgePublishPacket(ushort packetIdentifier)
         {
-            if (packet == null) throw new ArgumentNullException(nameof(packet));
+            _publishPacketsWithoutAcknowledge.Remove(packetIdentifier);
+        }
+        
+        public void EnqueuePacket(MqttPacketBusItem packetBusItem)
+        {
+            if (packetBusItem == null) throw new ArgumentNullException(nameof(packetBusItem));
 
             if (_packetBus.PacketsCount >= _serverOptions.MaxPendingMessagesPerClient)
             {
-                // TODO: Log!
+                if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
-                return;
+                {
+                    // TODO: Log!
+                    return;
+                }
+                else
+                {
+                    // TODO: Implement.
+                }
             }
             
-            if (packet.Packet is MqttPublishPacket)
+            if (packetBusItem.Packet is MqttPublishPacket publishPacket)
             {
-                _packetBus.Enqueue(packet, MqttPacketBusPartition.Data);    
+                if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+                {
+                    _publishPacketsWithoutAcknowledge[publishPacket.PacketIdentifier] = publishPacket;
+                }
+                
+                _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Data);
             }
-            else if (packet.Packet is MqttPingReqPacket || packet.Packet is MqttPingRespPacket)
+            else if (packetBusItem.Packet is MqttPingReqPacket || packetBusItem.Packet is MqttPingRespPacket)
             {
-                _packetBus.Enqueue(packet, MqttPacketBusPartition.Health);    
+                _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Health);    
             }
             else
             {
-                _packetBus.Enqueue(packet, MqttPacketBusPartition.Control);
+                _packetBus.Enqueue(packetBusItem, MqttPacketBusPartition.Control);
             }
         }
 
@@ -80,13 +102,25 @@ namespace MQTTnet.Server.Internal
         
         public Task DeleteAsync()
         {
-            return _clientSessionsManager.DeleteSessionAsync(ClientId);
+            return _clientSessionsManager.DeleteSessionAsync(Id);
         }
         
         public void OnDeleted()
         {
             Deleted?.Invoke(this, EventArgs.Empty);
         }
+        
+        public void Recover()
+        {
+            // TODO: Keep the bus and only insert pending items again.
+            // TODO: Check if packet identifier must be restarted or not.
+            _packetBus.Clear();
+
+            foreach (var publishPacket in _publishPacketsWithoutAcknowledge.Values.ToList())
+            {
+                EnqueuePacket(new MqttPacketBusItem(publishPacket));
+            }
+        }
 
         public void Dispose()
         {

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -19,7 +19,8 @@ namespace MQTTnet.Server.Internal
     public sealed class MqttClientSessionsManager : IDisposable
     {
         readonly BlockingCollection<MqttPendingApplicationMessage> _messageQueue = new BlockingCollection<MqttPendingApplicationMessage>();
-
+        readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
+        
         readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
         readonly Dictionary<string, MqttClientConnection> _clientConnections = new Dictionary<string, MqttClientConnection>(4096);
         readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>(4096);
@@ -129,36 +130,39 @@ namespace MQTTnet.Server.Internal
             {
                 if (clientConnection != null)
                 {
-                    if (clientConnection.ClientId != null)
+                    if (clientConnection.Id != null)
                     {
                         // in case it is a takeover _clientConnections already contains the new connection
                         if (!clientConnection.IsTakenOver)
                         {
                             lock (_clientConnections)
                             {
-                                _clientConnections.Remove(clientConnection.ClientId);
+                                _clientConnections.Remove(clientConnection.Id);
                             }
 
                             if (!_options.EnablePersistentSessions)
                             {
-                                await DeleteSessionAsync(clientConnection.ClientId).ConfigureAwait(false);
+                                await DeleteSessionAsync(clientConnection.Id).ConfigureAwait(false);
                             }
                         }
                     }
 
                     var endpoint = clientConnection.Endpoint;
 
-                    if (clientConnection.ClientId != null && !clientConnection.IsTakenOver)
+                    if (clientConnection.Id != null && !clientConnection.IsTakenOver)
                     {
                         // The event is fired at a separate place in case of a handover!
                         await _eventDispatcher.SafeNotifyClientDisconnectedAsync(
-                            clientConnection.ClientId,
+                            clientConnection.Id,
                             clientConnection.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean,
                             endpoint).ConfigureAwait(false);
                     }
                 }
 
-                await channelAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
+                using (var timeout = new CancellationTokenSource(_options.DefaultCommunicationTimeout))
+                {
+                    await channelAdapter.DisconnectAsync(timeout.Token).ConfigureAwait(false);    
+                }
             }
         }
 
@@ -221,11 +225,11 @@ namespace MQTTnet.Server.Internal
             return Task.FromResult((IList<IMqttSessionStatus>)result);
         }
 
-        public void DispatchApplicationMessage(MqttApplicationMessage applicationMessage, MqttClientConnection sender)
+        public void DispatchPublishPacket(MqttPublishPacket publishPacket, MqttClientConnection sender)
         {
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            _messageQueue.Add(new MqttPendingApplicationMessage(applicationMessage, sender));
+            _messageQueue.Add(new MqttPendingApplicationMessage(publishPacket, sender));
         }
 
         public async Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
@@ -242,11 +246,7 @@ namespace MQTTnet.Server.Internal
             
             foreach (var retainedApplicationMessage in subscribeResult.RetainedApplicationMessages)
             {
-                //clientSession.ApplicationMessagesQueue.Enqueue(retainedApplicationMessage);
+                var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage);
-
-                // TODO: Review!
-                var publishPacketFactory = new MqttPublishPacketFactory();
-                var publishPacket = publishPacketFactory.Create(retainedApplicationMessage.ApplicationMessage);
                 clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket));
             }
         }
@@ -351,8 +351,10 @@ namespace MQTTnet.Server.Internal
                 }
 
                 var clientConnection = pendingApplicationMessage.Sender;
-                var senderClientId = clientConnection?.ClientId ?? _options.ClientId;
+                var senderClientId = clientConnection?.Id ?? _options.ClientId;
-                var applicationMessage = pendingApplicationMessage.ApplicationMessage;
+                
+                // TODO: Only use PUBLISH packets internally!
+                var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket);
 
                 var interceptor = _options.ApplicationMessageInterceptor;
                 if (interceptor != null)
@@ -403,7 +405,7 @@ namespace MQTTnet.Server.Internal
                         continue;
                     }
 
-                    _logger.Verbose("Client '{0}': Queued application message with topic '{1}'.", clientSession.ClientId, applicationMessage.Topic);
+                    _logger.Verbose("Client '{0}': Queued application message with topic '{1}'.", clientSession.Id, applicationMessage.Topic);
 
                     // var queuedApplicationMessage = new MqttQueuedApplicationMessage
                     // {
@@ -411,16 +413,12 @@ namespace MQTTnet.Server.Internal
                     //     SubscriptionQualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel,
                     //     SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers
                     // };
-                    
+                   
-
+                    var publishPacket = _packetFactories.Publish.Create(applicationMessage);
-
-                    var publishPacketFactory = new MqttPublishPacketFactory();
-                    var publishPacket = publishPacketFactory.Create(applicationMessage);
                     
                     publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
                     publishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers);
-                    //clientSession.ApplicationMessagesQueue.Enqueue(queuedApplicationMessage);
+
-                    
                     if (publishPacket.QualityOfServiceLevel > 0)
                     {
                         publishPacket.PacketIdentifier = clientSession.PacketIdentifierProvider.GetNextPacketIdentifier();
@@ -458,34 +456,6 @@ namespace MQTTnet.Server.Internal
                 _logger.Error(exception, "Unhandled exception while processing next queued application message.");
             }
         }
-        
-        // public async Task<MqttPublishPacket> CreatePublishPacket(MqttQueuedApplicationMessage queuedApplicationMessage)
-        // {
-        //     var publishPacket = ChannelAdapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(queuedApplicationMessage.ApplicationMessage);
-        //     publishPacket.QualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel;
-        //
-        //     if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
-        //     {
-        //         publishPacket.Properties.SubscriptionIdentifiers.AddRange(queuedApplicationMessage.SubscriptionIdentifiers ?? new List<uint>());
-        //     }
-        //
-        //     // Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9].
-        //     publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage;
-        //
-        //     publishPacket = await InvokeClientMessageQueueInterceptor(publishPacket, queuedApplicationMessage).ConfigureAwait(false);
-        //     if (publishPacket == null)
-        //     {
-        //         // The interceptor has decided that the message is not relevant and will be fully ignored.
-        //         return null;
-        //     }
-        //
-        //     if (publishPacket.QualityOfServiceLevel > 0)
-        //     {
-        //         publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
-        //     }
-        //
-        //     return publishPacket;
-        // }
 
         async Task<MqttConnectionValidatorContext> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
         {
@@ -547,6 +517,7 @@ namespace MQTTnet.Server.Internal
                         {
                             _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
                             connAckPacket.IsSessionPresent = true;
+                            session.Recover();
                         }
                     }
 
@@ -565,7 +536,7 @@ namespace MQTTnet.Server.Internal
 
                 if (existingConnection != null)
                 {
-                    await _eventDispatcher.SafeNotifyClientDisconnectedAsync(existingConnection.ClientId, MqttClientDisconnectType.Takeover, existingConnection.Endpoint);
+                    await _eventDispatcher.SafeNotifyClientDisconnectedAsync(existingConnection.Id, MqttClientDisconnectType.Takeover, existingConnection.Endpoint);
 
                     existingConnection.IsTakenOver = true;
                     await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
@@ -591,7 +562,7 @@ namespace MQTTnet.Server.Internal
             }
             else
             {
-                senderClientId = clientConnection.ClientId;
+                senderClientId = clientConnection.Id;
                 sessionItems = clientConnection.Session.Items;
             }
 
@@ -599,7 +570,7 @@ namespace MQTTnet.Server.Internal
             {
                 ClientId = senderClientId,
                 SessionItems = sessionItems,
-                AcceptPublish = true,
+                ProcessPublish = true,
                 ApplicationMessage = applicationMessage,
                 CloseConnection = false
             };

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -16,7 +16,7 @@ namespace MQTTnet.Server.Internal
         // completely by swapping references etc.
         readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim();
         readonly Dictionary<string, Subscription> _subscriptions = new Dictionary<string, Subscription>(4096);
-        
+
         readonly MqttClientSession _clientSession;
         readonly IMqttServerOptions _options;
         readonly MqttServerEventDispatcher _eventDispatcher;
@@ -67,7 +67,7 @@ namespace MQTTnet.Server.Internal
                     // must be closed. So do not revert to false!
                     result.CloseConnection = true;
                 }
-                
+
                 if (!acceptSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic))
                 {
                     continue;
@@ -81,7 +81,7 @@ namespace MQTTnet.Server.Internal
                         interceptorContext.ReasonCode);
 
                     await _eventDispatcher
-                        .SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter)
+                        .SafeNotifyClientSubscribedTopicAsync(_clientSession.Id, finalTopicFilter)
                         .ConfigureAwait(false);
 
                     FilterRetainedApplicationMessages(retainedApplicationMessages, subscription, result);
@@ -103,7 +103,7 @@ namespace MQTTnet.Server.Internal
                 foreach (var topicFilter in unsubscribePacket.TopicFilters)
                 {
                     _subscriptions.TryGetValue(topicFilter, out var existingSubscription);
-                    
+
                     var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription, cancellationToken).ConfigureAwait(false);
                     var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success;
 
@@ -115,7 +115,7 @@ namespace MQTTnet.Server.Internal
                         // must be closed. So do not revert to false!
                         result.CloseConnection = true;
                     }
-           
+
                     if (!acceptUnsubscription)
                     {
                         continue;
@@ -134,7 +134,7 @@ namespace MQTTnet.Server.Internal
 
             foreach (var topicFilter in unsubscribePacket.TopicFilters)
             {
-                await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.Id, topicFilter).ConfigureAwait(false);
             }
 
             return result;
@@ -154,7 +154,7 @@ namespace MQTTnet.Server.Internal
                 _subscriptionsLock.ExitReadLock();
             }
 
-            var senderIsReceiver = string.Equals(senderClientId, _clientSession.ClientId);
+            var senderIsReceiver = string.Equals(senderClientId, _clientSession.Id);
 
             var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
             var subscriptionIdentifiers = new HashSet<uint>();
@@ -296,7 +296,7 @@ namespace MQTTnet.Server.Internal
         {
             var context = new MqttSubscriptionInterceptorContext
             {
-                ClientId = _clientSession.ClientId,
+                ClientId = _clientSession.Id,
                 TopicFilter = topicFilter,
                 SessionItems = _clientSession.Items,
                 Session = new MqttSessionStatus(_clientSession),
@@ -343,7 +343,7 @@ namespace MQTTnet.Server.Internal
         {
             var context = new MqttUnsubscriptionInterceptorContext
             {
-                ClientId = _clientSession.ClientId,
+                ClientId = _clientSession.Id,
                 Topic = topicFilter,
                 SessionItems = _clientSession.Items,
                 CancellationToken = cancellationToken
@@ -369,8 +369,7 @@ namespace MQTTnet.Server.Internal
             return context;
         }
 
-        static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel,
+        static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel, ICollection<MqttQualityOfServiceLevel> subscribedQoSLevels)
-            HashSet<MqttQualityOfServiceLevel> subscribedQoSLevels)
         {
             MqttQualityOfServiceLevel effectiveQoS;
             if (subscribedQoSLevels.Contains(qosLevel))

Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs

@@ -101,7 +101,7 @@ namespace MQTTnet.Server.Internal
                     return;
                 }
 
-                _logger.Warning("Client '{0}': Did not receive any packet or keep alive signal.", connection.ClientId);
+                _logger.Warning("Client '{0}': Did not receive any packet or keep alive signal.", connection.Id);
 
                 // Execute the disconnection in background so that the keep alive monitor can continue
                 // with checking other connections.
@@ -112,7 +112,7 @@ namespace MQTTnet.Server.Internal
             }
             catch (Exception exception)
             {
-                _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.ClientId);
+                _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.Id);
             }
         }
     }

Source/MQTTnet/Server/MqttConnectionValidatorContext.cs

@@ -41,12 +41,6 @@ namespace MQTTnet.Server
 
         public string Password => Encoding.UTF8.GetString(RawPassword ?? PlatformAbstractionLayer.EmptyByteArray);
 
-        /// <summary>
-        /// Gets or sets the will delay interval.
-        /// This is the time between the client disconnect and the time the will message will be sent.
-        /// </summary>
-        public MqttApplicationMessage WillMessage => _connectPacket.WillMessage;
-
         /// <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 persistent connection.
@@ -123,7 +117,7 @@ namespace MQTTnet.Server
         /// Gets or sets the will delay interval.
         /// This is the time between the client disconnect and the time the will message will be sent.
         /// </summary>
-        public uint? WillDelayInterval => _connectPacket.Properties?.WillDelayInterval;
+        public uint? WillDelayInterval => _connectPacket.WillProperties?.WillDelayInterval;
 
         /// <summary>
         /// Gets or sets a key/value collection that can be used to share data within the scope of this session.

Source/MQTTnet/Server/MqttPendingApplicationMessage.cs

@@ -1,17 +1,18 @@
-using MQTTnet.Server.Internal;
+using MQTTnet.Packets;
+using MQTTnet.Server.Internal;
 
 namespace MQTTnet.Server
 {
     public sealed class MqttPendingApplicationMessage
     {
-        public MqttPendingApplicationMessage(MqttApplicationMessage applicationMessage, MqttClientConnection sender)
+        public MqttPendingApplicationMessage(MqttPublishPacket publishPacket, MqttClientConnection sender)
         {
             Sender = sender;
-            ApplicationMessage = applicationMessage;
+            PublishPacket = publishPacket;
         }
 
         public MqttClientConnection Sender { get; }
 
-        public MqttApplicationMessage ApplicationMessage { get; }
+        public MqttPublishPacket PublishPacket { get; }
     }
 }

Source/MQTTnet/Server/MqttServer.cs

@@ -1,7 +1,6 @@
 using MQTTnet.Adapter;
 using MQTTnet.Client.Publishing;
 using MQTTnet.Client.Receiving;
-using MQTTnet.Diagnostics;
 using MQTTnet.Exceptions;
 using MQTTnet.Protocol;
 using MQTTnet.Server.Status;
@@ -11,6 +10,7 @@ using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
+using MQTTnet.Formatter;
 using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.Server.Internal;
@@ -19,6 +19,7 @@ namespace MQTTnet.Server
 {
     public class MqttServer : Disposable, IMqttServer
     {
+        readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
         readonly MqttServerEventDispatcher _eventDispatcher;
         readonly ICollection<IMqttServerAdapter> _adapters;
         readonly IMqttNetLogger _rootLogger;
@@ -148,7 +149,8 @@ namespace MQTTnet.Server
 
             ThrowIfNotStarted();
 
-            _clientSessionsManager.DispatchApplicationMessage(applicationMessage, null);
+            var publishPacket = _packetFactories.Publish.Create(applicationMessage);
+            _clientSessionsManager.DispatchPublishPacket(publishPacket, null);
 
             return Task.FromResult(new MqttClientPublishResult());
         }

Source/MQTTnet/Server/Status/MqttClientStatus.cs

@@ -19,7 +19,7 @@ namespace MQTTnet.Server.Status
         /// Gets or sets 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 => _connection.ClientId;
+        public string ClientId => _connection.Id;
 
         public string Endpoint => _connection.Endpoint;
 

Source/MQTTnet/Server/Status/MqttSessionStatus.cs

@@ -24,10 +24,9 @@ namespace MQTTnet.Server.Status
             remove => _session.Deleted += value;
         }
 
-        public string ClientId => _session.ClientId;
+        public string ClientId => _session.Id;
 
-        // TODO: Fix!
+        public long PendingApplicationMessagesCount => _session.PendingDataPacketsCount;
-        public long PendingApplicationMessagesCount => 0; // _session.ApplicationMessagesQueue.Count;
 
         public DateTime CreatedTimestamp => _session.CreatedTimestamp;
 

Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs

@@ -23,10 +23,8 @@ using MQTTnet.Tests.Mockups;
 namespace MQTTnet.Tests.Client
 {
     [TestClass]
-    public class Client_Tests
+    public class Client_Tests : BaseTestClass
     {
-        public TestContext TestContext { get; set; }
-
         [TestMethod]
         public async Task Ensure_Queue_Drain()
         {
@@ -690,7 +688,7 @@ namespace MQTTnet.Tests.Client
         [TestMethod]
         public async Task Message_Send_Retry()
         {
-            using (var testEnvironment = new TestEnvironment(TestContext))
+            using (var testEnvironment = CreateTestEnvironment())
             {
                 testEnvironment.IgnoreClientLogErrors = true;
                 testEnvironment.IgnoreServerLogErrors = true;
@@ -914,40 +912,5 @@ namespace MQTTnet.Tests.Client
                 Assert.IsTrue(success);
             }
         }
-
-        [DataTestMethod]
-        [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
-        [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
-        public async Task Retry_If_Not_PubAck(MqttQualityOfServiceLevel qos)
-        {
-            long count = 0;
-            using (var testEnvironment = new TestEnvironment(TestContext))
-            {
-                await testEnvironment.StartServer();
-                var publisher = await testEnvironment.ConnectClient();
-                var subscriber = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId(qos.ToString()));
-                await subscriber.SubscribeAsync("#", qos);
-
-                subscriber.UseApplicationMessageReceivedHandler(c =>
-                {
-                    c.AutoAcknowledge = false;
-                    ++count;
-                    Console.WriteLine("process");
-                    return Task.CompletedTask;
-                });
-
-                var pub = publisher.PublishAsync("a", null, qos);
-
-                await Task.Delay(100);
-                await subscriber.DisconnectAsync();
-                await subscriber.ReconnectAsync();
-                await Task.Delay(100);
-
-                var res = await pub;
-
-                Assert.AreEqual(MqttClientPublishReasonCode.Success, res.ReasonCode);
-                Assert.AreEqual(2, count);
-            }
-        }
     }
 }

Tests/MQTTnet.Core.Tests/Extension_Tests.cs

@@ -27,16 +27,7 @@ namespace MQTTnet.Tests
                 ct => Task.Delay(TimeSpan.FromMilliseconds(500), ct).ContinueWith(t => 5, ct),
                 TimeSpan.FromMilliseconds(100), CancellationToken.None);
         }
-
+        
-        [TestMethod]
-        public async Task TimeoutAfterCompleteInTime()
-        {
-            var result = await MqttTaskTimeout.WaitAsync(
-                ct => Task.Delay(TimeSpan.FromMilliseconds(100), ct).ContinueWith(t => 5, ct),
-                TimeSpan.FromMilliseconds(500), CancellationToken.None);
-            Assert.AreEqual(5, result);
-        }
-
         [TestMethod]
         public async Task TimeoutAfterWithInnerException()
         {

Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs

@@ -118,7 +118,7 @@ namespace MQTTnet.Tests.MQTTv5
                 await client.DisconnectAsync();
 
                 Assert.AreEqual(1, result.Items.Count);
-                Assert.AreEqual(MqttClientUnsubscribeResultCode.Success, result.Items[0].ReasonCode);
+                Assert.AreEqual(MqttClientUnsubscribeResultCode.Success, result.Items[0].ResultCode);
             }
         }
 

Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs

@@ -37,7 +37,7 @@ namespace MQTTnet.Tests.Mockups
             return Task.FromResult(0);
         }
 
-        public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+        public Task DisconnectAsync(CancellationToken cancellationToken)
         {
             return Task.FromResult(0);
         }

Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs

@@ -9,7 +9,6 @@ using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using MQTTnet.Formatter.V3;
-using MQTTnet.Formatter.V5;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
@@ -94,13 +93,11 @@ namespace MQTTnet.Tests
                 Username = "USER",
                 KeepAlivePeriod = 123,
                 CleanSession = true,
-                WillMessage = new MqttApplicationMessage
+                WillFlag = true,
-                {
+                WillTopic = "My/last/will",
-                    Topic = "My/last/will",
+                WillMessage = Encoding.UTF8.GetBytes("Good byte."),
-                    Payload = Encoding.UTF8.GetBytes("Good byte."),
+                WillQoS =  MqttQualityOfServiceLevel.AtLeastOnce,
-                    QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
+                WillRetain = true
-                    Retain = true
-                }
             };
 
             SerializeAndCompare(p, "EDUABE1RVFQE7gB7AANYWVoADE15L2xhc3Qvd2lsbAAKR29vZCBieXRlLgAEVVNFUgAEUEFTUw==");
@@ -131,13 +128,11 @@ namespace MQTTnet.Tests
                 Username = "USER",
                 KeepAlivePeriod = 123,
                 CleanSession = true,
-                WillMessage = new MqttApplicationMessage
+                WillFlag = true,
-                {
+                WillTopic = "My/last/will",
-                    Topic = "My/last/will",
+                WillMessage = Encoding.UTF8.GetBytes("Good byte."),
-                    Payload = Encoding.UTF8.GetBytes("Good byte."),
+                WillQoS =  MqttQualityOfServiceLevel.AtLeastOnce,
-                    QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
+                WillRetain = true
-                    Retain = true
-                }
             };
 
             DeserializeAndCompare(p, "EDUABE1RVFQE7gB7AANYWVoADE15L2xhc3Qvd2lsbAAKR29vZCBieXRlLgAEVVNFUgAEUEFTUw==");

Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs

@@ -5,6 +5,7 @@ using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Client;
 using MQTTnet.Client.Options;
+using MQTTnet.Client.Publishing;
 using MQTTnet.Client.Subscribing;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
@@ -165,6 +166,46 @@ namespace MQTTnet.Tests.Server
                 Assert.AreEqual("injected_one", messageReceivedHandler.ReceivedApplicationMessages[0].Topic);
             }
         }
+        
+        [DataTestMethod]
+        [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
+        [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
+        public async Task Retry_If_Not_PubAck(MqttQualityOfServiceLevel qos)
+        {
+            long count = 0;
+            using (var testEnvironment = CreateTestEnvironment())
+            {
+                await testEnvironment.StartServer(o => o.WithPersistentSessions());
+                
+                var publisher = await testEnvironment.ConnectClient();
+                
+                var subscriber = await testEnvironment.ConnectClient(o => o
+                    .WithClientId(qos.ToString())
+                    .WithCleanSession(false));
+                
+                subscriber.UseApplicationMessageReceivedHandler(c =>
+                {
+                    c.AutoAcknowledge = false;
+                    ++count;
+                    Console.WriteLine("process");
+                    return Task.CompletedTask;
+                });
+                
+                await subscriber.SubscribeAsync("#", qos);
+
+                var pub = publisher.PublishAsync("a", null, qos);
+
+                await Task.Delay(100);
+                await subscriber.DisconnectAsync();
+                await subscriber.ReconnectAsync();
+                await Task.Delay(100);
+
+                var res = await pub;
+
+                Assert.AreEqual(MqttClientPublishReasonCode.Success, res.ReasonCode);
+                Assert.AreEqual(2, count);
+            }
+        }
 
         async Task<IMqttClient> TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options)
         {

Tests/MQTTnet.TestApp.NetCore/ServerTest.cs

@@ -64,7 +64,7 @@ namespace MQTTnet.TestApp.NetCore
 
                         if (context.ApplicationMessage.Topic == "not_allowed_topic")
                         {
-                            context.AcceptPublish = false;
+                            context.ProcessPublish = false;
                             context.CloseConnection = true;
                         }
                     }),