Accept multiple SubscriptionIdentifiers for PUBLISH packets.

Build/MQTTnet.Extensions.ManagedClient.nuspec

@@ -34,10 +34,9 @@
     <!-- .NET Framework -->
     <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\net452\MQTTnet.Extensions.ManagedClient.*" target="lib\net452\"/>
 
-    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard2.0\MQTTnet.Extensions.ManagedClient.*" target="lib\net461\"/>
-    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard2.0\MQTTnet.Extensions.ManagedClient.*" target="lib\net462\"/>
-    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard2.0\MQTTnet.Extensions.ManagedClient.*" target="lib\net470\"/>
-    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard2.0\MQTTnet.Extensions.ManagedClient.*" target="lib\net471\"/>
-    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\netstandard2.0\MQTTnet.Extensions.ManagedClient.*" target="lib\net472\"/>
+    <!-- .NET Framework 4.6.0 will use binaries from 4.5.2. -->
+    <file src="..\Source\MQTTnet.Extensions.ManagedClient\bin\Release\net461\MQTTnet.Extensions.Rpc.*" target="lib\net461\"/>
+
+    <!-- Everything above .NET Framework 4.6.1 will use the binaries from 4.6.1. -->
   </files>
 </package>

Build/MQTTnet.Extensions.Rpc.nuspec

@@ -34,10 +34,9 @@
     <!-- .NET Framework -->
     <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\net452\MQTTnet.Extensions.Rpc.*" target="lib\net452\"/>
 
-    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard2.0\MQTTnet.Extensions.Rpc.*" target="lib\net461\"/>
-    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard2.0\MQTTnet.Extensions.Rpc.*" target="lib\net462\"/>
-    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard2.0\MQTTnet.Extensions.Rpc.*" target="lib\net470\"/>
-    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard2.0\MQTTnet.Extensions.Rpc.*" target="lib\net471\"/>
-    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\netstandard2.0\MQTTnet.Extensions.Rpc.*" target="lib\net472\"/>
+    <!-- .NET Framework 4.6.0 will use binaries from 4.5.2. -->
+    <file src="..\Source\MQTTnet.Extensions.Rpc\bin\Release\net461\MQTTnet.Extensions.Rpc.*" target="lib\net461\"/>
+
+    <!-- Everything above .NET Framework 4.6.1 will use the binaries from 4.6.1. -->
   </files>
 </package>

Build/MQTTnet.nuspec

@@ -11,36 +11,7 @@
     <requireLicenseAcceptance>false</requireLicenseAcceptance>
     <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
     <releaseNotes>
-* [Core] Added support for MQTTv5 packages.
-* [Core] Performance improvements.
-* [Core] Removed obsolete methods.
-* [Core] Fixed a memory leak when processing lots of messages (thanks to @tschanko)
-* [Core] Added more overloads for MQTT factory.
-* [Core] The client password is now hidden from the logs (replaced with **** if set).
-* [Core] Fixed a memory leak when using SSL connections (thanks to @biovoid). 
-* [Client] Added validation of topics before publishing.
-* [Client] Added new MQTTv5 features to options builder.
-* [Client] Added uniform API across all supported MQTT versions (BREAKING CHANGE!)
-* [Client] The client will now avoid sending an ACK if an exception has been thrown in message handler (thanks to @ramonsmits).
-* [Client] Fixed issues in QoS 2 handling which leads to message loss.
-* [Client] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
-* [Client] Added more configuration values to TCP endpoint options.
-* [Client] Added used PacketIdentifier to publish result.
-* [ManagedClient] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
-* [ManagedClient] The log ID is now propagated to the internal client (thanks to @vbBerni).
-* [ManagedClient] Added validation of topics before publishing.
-* [ManagedClient] The internal MQTT client is now closed properly (thanks to @vbBerni).
-* [Server] Added support for MQTTv5 clients. The server will still return _success_ for all cases at the moment even if more granular codes are available.
-* [Server] Fixed issues in QoS 2 handling which leads to message loss.
-* [Server] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
-* [Server] The used logger instance is now propagated to the WebSocket server adapter.
-* [Server] Added the flag "IsSecureConnection" which is set to true when the connection is encrypted.
-* [Server] Fixed wrong will message behavior when stopping server (thanks to @JohBa)
-* [Server] Added validation of topics before publishing.
-* [Server] Added more configuration values to TCP endpoint options.
-* [MQTTnet Server] Added as first Alpha version of standalone cross platform MQTT server.
-
-* [Note] Due to MQTTv5 a lot of new classes were introduced. This required adding new namespaces as well. Most classes are backward compatible but new namespaces must be added.
+* [Core] Fixed issues in MQTTv5 message encoding and decoding.
     </releaseNotes>
     <copyright>Copyright Christian Kratky 2016-2019</copyright>
     <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> 
@@ -60,10 +31,6 @@
       <group targetFramework="uap10.0">
         <dependency id="Microsoft.NETCore.UniversalWindowsPlatform" version="6.2.8" />
       </group>
-      <group targetFramework="net452">
-      </group>
-      <group targetFramework="net461">
-      </group>
     </dependencies>
   </metadata>
 
@@ -83,10 +50,9 @@
     <!-- .NET Framework -->
     <file src="..\Source\MQTTnet\bin\Release\net452\MQTTnet.*" target="lib\net452\"/>
 
-    <file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\net461\"/>
-    <file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\net462\"/>
-    <file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\net470\"/>
-    <file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\net471\"/>
-    <file src="..\Source\MQTTnet\bin\Release\netstandard2.0\MQTTnet.*" target="lib\net472\"/>
+    <!-- .NET Framework 4.6.0 will use binaries from 4.5.2. -->
+    <file src="..\Source\MQTTnet\bin\Release\net461\MQTTnet.*" target="lib\net461\"/>
+
+    <!-- Everything above .NET Framework 4.6.1 will use the binaries from 4.6.1. -->
   </files>
 </package>

Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs

@@ -1,17 +1,21 @@
 using System;
+using MQTTnet.Client.Connecting;
 
 namespace MQTTnet.Client.Disconnecting
 {
     public class MqttClientDisconnectedEventArgs : EventArgs
     {
-        public MqttClientDisconnectedEventArgs(bool clientWasConnected, Exception exception)
+        public MqttClientDisconnectedEventArgs(bool clientWasConnected, Exception exception, MqttClientAuthenticateResult authenticateResult)
         {
             ClientWasConnected = clientWasConnected;
             Exception = exception;
+            AuthenticateResult = authenticateResult;
         }
 
         public bool ClientWasConnected { get; }
 
         public Exception Exception { get; }
+
+        public MqttClientAuthenticateResult AuthenticateResult { get; }
     }
 }

Source/MQTTnet/Client/MqttClient.cs

@@ -61,6 +61,8 @@ namespace MQTTnet.Client
 
             ThrowIfConnected("It is not allowed to connect with a server after the connection is established.");
 
+            MqttClientAuthenticateResult authenticateResult = null;
+
             try
             {
                 Options = options;
@@ -81,7 +83,7 @@ namespace MQTTnet.Client
 
                 _packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken);
 
-                var authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, cancellationToken).ConfigureAwait(false);
+                authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, cancellationToken).ConfigureAwait(false);
 
                 _sendTracker.Restart();
 
@@ -108,7 +110,7 @@ namespace MQTTnet.Client
 
                 if (!DisconnectIsPending())
                 {
-                    await DisconnectInternalAsync(null, exception).ConfigureAwait(false);
+                    await DisconnectInternalAsync(null, exception, authenticateResult).ConfigureAwait(false);
                 }
 
                 throw;
@@ -131,7 +133,7 @@ namespace MQTTnet.Client
             {
                 if (!DisconnectIsPending())
                 {
-                    await DisconnectInternalAsync(null, null).ConfigureAwait(false);
+                    await DisconnectInternalAsync(null, null, null).ConfigureAwait(false);
                 }
             }
         }
@@ -232,7 +234,7 @@ namespace MQTTnet.Client
             if (IsConnected) throw new MqttProtocolViolationException(message);
         }
 
-        private async Task DisconnectInternalAsync(Task sender, Exception exception)
+        private async Task DisconnectInternalAsync(Task sender, Exception exception, MqttClientAuthenticateResult authenticateResult)
         {
             var clientWasConnected = IsConnected;
 
@@ -267,7 +269,7 @@ namespace MQTTnet.Client
                 var disconnectedHandler = DisconnectedHandler;
                 if (disconnectedHandler != null)
                 {
-                    await disconnectedHandler.HandleDisconnectedAsync(new MqttClientDisconnectedEventArgs(clientWasConnected, exception)).ConfigureAwait(false);
+                    await disconnectedHandler.HandleDisconnectedAsync(new MqttClientDisconnectedEventArgs(clientWasConnected, exception, authenticateResult)).ConfigureAwait(false);
                 }
             }
         }
@@ -376,7 +378,7 @@ namespace MQTTnet.Client
 
                 if (!DisconnectIsPending())
                 {
-                    await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception).ConfigureAwait(false);
+                    await DisconnectInternalAsync(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
                 }
             }
             finally
@@ -404,7 +406,7 @@ namespace MQTTnet.Client
                     {
                         if (!DisconnectIsPending())
                         {
-                            await DisconnectInternalAsync(_packetReceiverTask, null).ConfigureAwait(false);
+                            await DisconnectInternalAsync(_packetReceiverTask, null, null).ConfigureAwait(false);
                         }
 
                         return;
@@ -436,7 +438,7 @@ namespace MQTTnet.Client
 
                 if (!DisconnectIsPending())
                 {
-                    await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false);
+                    await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
                 }
             }
             finally
@@ -467,6 +469,9 @@ namespace MQTTnet.Client
                 }
                 else if (packet is MqttDisconnectPacket)
                 {
+                    // Also dispatch disconnect to waiting threads to generate a proper exception.
+                    _packetDispatcher.Dispatch(packet);
+
                     await DisconnectAsync(null, cancellationToken).ConfigureAwait(false);
                 }
                 else
@@ -497,7 +502,7 @@ namespace MQTTnet.Client
 
                 if (!DisconnectIsPending())
                 {
-                    await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false);
+                    await DisconnectInternalAsync(_packetReceiverTask, exception, null).ConfigureAwait(false);
                 }
             }
         }

Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs

@@ -136,6 +136,16 @@ namespace MQTTnet.Client.Options
             return this;
         }
 
+        public MqttClientOptionsBuilder WithTcpServer(Action<MqttClientTcpOptions> optionsBuilder)
+        {
+            if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder));
+
+            _tcpOptions = new MqttClientTcpOptions();
+            optionsBuilder.Invoke(_tcpOptions);
+
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithProxy(string address, string username = null, string password = null, string domain = null, bool bypassOnLocal = false, string[] bypassList = null)
         {
             _proxyOptions = new MqttClientWebSocketProxyOptions
@@ -151,6 +161,15 @@ namespace MQTTnet.Client.Options
             return this;
         }
 
+        public MqttClientOptionsBuilder WithProxy(Action<MqttClientWebSocketProxyOptions> optionsBuilder)
+        {
+            if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder));
+
+            _proxyOptions = new MqttClientWebSocketProxyOptions();
+            optionsBuilder(_proxyOptions);
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithWebSocketServer(string uri, MqttClientOptionsBuilderWebSocketParameters parameters = null)
         {
             _webSocketOptions = new MqttClientWebSocketOptions
@@ -163,9 +182,19 @@ namespace MQTTnet.Client.Options
             return this;
         }
 
+        public MqttClientOptionsBuilder WithWebSocketServer(Action<MqttClientWebSocketOptions> optionsBuilder)
+        {
+            if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder));
+
+            _webSocketOptions = new MqttClientWebSocketOptions();
+            optionsBuilder.Invoke(_webSocketOptions);
+
+            return this;
+        }
+
         public MqttClientOptionsBuilder WithTls(MqttClientOptionsBuilderTlsParameters parameters)
         {
-            _tlsParameters = parameters ?? throw new ArgumentNullException(nameof(parameters));
+            _tlsParameters = parameters;
             return this;
         }
 
@@ -174,6 +203,15 @@ namespace MQTTnet.Client.Options
             return WithTls(new MqttClientOptionsBuilderTlsParameters { UseTls = true });
         }
 
+        public MqttClientOptionsBuilder WithTls(Action<MqttClientOptionsBuilderTlsParameters> optionsBuilder)
+        {
+            if (optionsBuilder == null) throw new ArgumentNullException(nameof(optionsBuilder));
+
+            _tlsParameters = new MqttClientOptionsBuilderTlsParameters();
+            optionsBuilder(_tlsParameters);
+            return this;
+        }
+
         public IMqttClientOptions Build()
         {
             if (_tcpOptions == null && _webSocketOptions == null)

Source/MQTTnet/Client/Subscribing/MqttClientSubscribeOptions.cs

@@ -7,6 +7,8 @@ namespace MQTTnet.Client.Subscribing
     {
         public List<TopicFilter> TopicFilters { get; set; } = new List<TopicFilter>();
 
+        public uint? SubscriptionIdentifier { get; set; }
+
         public List<MqttUserProperty> UserProperties { get; set; } = new List<MqttUserProperty>();
     }
 }

Source/MQTTnet/Exceptions/MqttUnexpectedDisconnectReceivedException.cs

@@ -0,0 +1,29 @@
+using System.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Exceptions
+{
+    public class MqttUnexpectedDisconnectReceivedException : MqttCommunicationException
+    {
+        public MqttUnexpectedDisconnectReceivedException(MqttDisconnectPacket disconnectPacket) 
+            : base($"Unexpected DISCONNECT (Reason code={disconnectPacket.ReasonCode}) received.")
+        {
+            ReasonCode = disconnectPacket.ReasonCode;
+            SessionExpiryInterval = disconnectPacket.Properties?.SessionExpiryInterval;
+            ReasonString = disconnectPacket.Properties?.ReasonString;
+            ServerReference = disconnectPacket.Properties?.ServerReference;
+            UserProperties = disconnectPacket.Properties?.UserProperties;
+        }
+
+        public MqttDisconnectReasonCode? ReasonCode { get; }
+
+        public uint? SessionExpiryInterval { get; }
+
+        public string ReasonString { get; }
+
+        public List<MqttUserProperty> UserProperties { get; }
+
+        public string ServerReference { get; }
+    }
+}

Source/MQTTnet/Formatter/IMqttPacketWriter.cs

@@ -8,7 +8,6 @@
         void Write(byte returnCode);
         void WriteWithLengthPrefix(byte[] payload);
         void Write(ushort keepAlivePeriod);
-
         void Write(IMqttPacketWriter propertyWriter);
         void WriteVariableLengthInteger(uint length);
         void Write(byte[] payload, int v, int length);

Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs

@@ -216,6 +216,7 @@ namespace MQTTnet.Formatter.V3
 
             return new MqttClientPublishResult
             {
+                PacketIdentifier = pubCompPacket.PacketIdentifier,
                 ReasonCode = MqttClientPublishReasonCode.Success
             };
         }

Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs

@@ -33,13 +33,14 @@ namespace MQTTnet.Formatter.V5
                     MessageExpiryInterval = applicationMessage.MessageExpiryInterval,
                     PayloadFormatIndicator = applicationMessage.PayloadFormatIndicator,
                     ResponseTopic = applicationMessage.ResponseTopic,
-                    SubscriptionIdentifier = applicationMessage.SubscriptionIdentifier,
+                    SubscriptionIdentifiers = applicationMessage.SubscriptionIdentifiers,
                     TopicAlias = applicationMessage.TopicAlias
                 }
             };
 
             if (applicationMessage.UserProperties != null)
             {
+                packet.Properties.UserProperties = new List<MqttUserProperty>();
                 packet.Properties.UserProperties.AddRange(applicationMessage.UserProperties);
             }
 
@@ -63,6 +64,13 @@ namespace MQTTnet.Formatter.V5
                 Payload = publishPacket.Payload,
                 QualityOfServiceLevel = publishPacket.QualityOfServiceLevel,
                 Retain = publishPacket.Retain,
+                ResponseTopic = publishPacket.Properties?.ResponseTopic,
+                ContentType = publishPacket.Properties?.ContentType,
+                CorrelationData = publishPacket.Properties?.CorrelationData,
+                MessageExpiryInterval = publishPacket.Properties?.MessageExpiryInterval,
+                SubscriptionIdentifiers = publishPacket.Properties?.SubscriptionIdentifiers,
+                TopicAlias = publishPacket.Properties?.TopicAlias,
+                PayloadFormatIndicator = publishPacket.Properties?.PayloadFormatIndicator,
                 UserProperties = publishPacket.Properties?.UserProperties ?? new List<MqttUserProperty>()
             };
         }
@@ -151,6 +159,7 @@ namespace MQTTnet.Formatter.V5
             };
 
             packet.TopicFilters.AddRange(options.TopicFilters);
+            packet.Properties.SubscriptionIdentifier = options.SubscriptionIdentifier;
             packet.Properties.UserProperties.AddRange(options.UserProperties);
 
             return packet;
@@ -223,6 +232,7 @@ namespace MQTTnet.Formatter.V5
             {
                 return new MqttClientPublishResult
                 {
+                    PacketIdentifier = pubCompPacket.PacketIdentifier,
                     ReasonCode = MqttClientPublishReasonCode.UnspecifiedError,
                     ReasonString = pubCompPacket.Properties?.ReasonString,
                     UserProperties = pubCompPacket.Properties?.UserProperties
@@ -231,6 +241,7 @@ namespace MQTTnet.Formatter.V5
 
             var result = new MqttClientPublishResult
             {
+                PacketIdentifier = pubCompPacket.PacketIdentifier,
                 ReasonCode = MqttClientPublishReasonCode.Success,
                 ReasonString = pubCompPacket.Properties?.ReasonString,
                 UserProperties = pubCompPacket.Properties?.UserProperties

Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using MQTTnet.Adapter;
 using MQTTnet.Exceptions;
@@ -91,10 +92,6 @@ namespace MQTTnet.Formatter.V5
                 {
                     packet.Properties.AuthenticationMethod = propertiesReader.ReadAuthenticationMethod();
                 }
-                else if (propertiesReader.CurrentPropertyId == MqttPropertyId.WillDelayInterval)
-                {
-                    packet.Properties.WillDelayInterval = propertiesReader.ReadWillDelayInterval();
-                }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AuthenticationData)
                 {
                     packet.Properties.AuthenticationData = propertiesReader.ReadAuthenticationData();
@@ -121,11 +118,11 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
-                    propertiesReader.ThrowInvalidPropertyIdException(typeof(MqttConnAckPacket));
+                    propertiesReader.ThrowInvalidPropertyIdException(typeof(MqttConnectPacket));
                 }
             }
 
@@ -133,6 +130,63 @@ namespace MQTTnet.Formatter.V5
 
             if (packet.WillMessage != null)
             {
+                var willPropertiesReader = new MqttV500PropertiesReader(body);
+
+                while (willPropertiesReader.MoveNext())
+                {
+                    if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.PayloadFormatIndicator)
+                    {
+                        packet.WillMessage.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator();
+                    }
+                    else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.MessageExpiryInterval)
+                    {
+                        packet.WillMessage.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();
+                    }
+                    else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.CorrelationData)
+                    {
+                        packet.WillMessage.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();
+                    }
+                    else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.WillDelayInterval)
+                    {
+                        // This is a special case!
+                        packet.Properties.WillDelayInterval = propertiesReader.ReadWillDelayInterval();
+                    }
+                    else if (willPropertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
+                    {
+                        if (packet.WillMessage.UserProperties == null)
+                        {
+                            packet.WillMessage.UserProperties = new List<MqttUserProperty>();
+                        }
+
+                        propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
+                    }
+                    else
+                    {
+                        propertiesReader.ThrowInvalidPropertyIdException(typeof(MqttPublishPacket));
+                    }
+                }
+
                 packet.WillMessage.Topic = body.ReadStringWithLengthPrefix();
                 packet.WillMessage.Payload = body.ReadWithLengthPrefix();
             }
@@ -186,7 +240,7 @@ namespace MQTTnet.Formatter.V5
                 {
                     packet.Properties.ReceiveMaximum = propertiesReader.ReadReceiveMaximum();
                 }
-                else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AssignedClientIdentifer)
+                else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AssignedClientIdentifier)
                 {
                     packet.Properties.AssignedClientIdentifier = propertiesReader.ReadAssignedClientIdentifier();
                 }
@@ -228,7 +282,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -266,7 +320,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -296,7 +350,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -346,7 +400,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -378,7 +432,7 @@ namespace MQTTnet.Formatter.V5
             {
                 if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -413,7 +467,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -487,7 +541,12 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier)
                 {
-                    packet.Properties.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier();
+                    if (packet.Properties.SubscriptionIdentifiers == null)
+                    {
+                        packet.Properties.SubscriptionIdentifiers = new List<uint>();
+                    }
+
+                    packet.Properties.SubscriptionIdentifiers.Add(propertiesReader.ReadSubscriptionIdentifier());
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.ContentType)
                 {
@@ -495,7 +554,12 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    if (packet.Properties.UserProperties == null)
+                    {
+                        packet.Properties.UserProperties = new List<MqttUserProperty>();
+                    }
+
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -538,7 +602,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -576,7 +640,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -614,7 +678,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -652,7 +716,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {
@@ -697,7 +761,7 @@ namespace MQTTnet.Formatter.V5
                 }
                 else if (propertiesReader.CurrentPropertyId == MqttPropertyId.UserProperty)
                 {
-                    propertiesReader.FillUserProperties(packet.Properties.UserProperties);
+                    propertiesReader.AddUserPropertyTo(packet.Properties.UserProperties);
                 }
                 else
                 {

Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs

@@ -127,24 +127,38 @@ namespace MQTTnet.Formatter.V5
             var propertiesWriter = new MqttV500PropertiesWriter();
             if (packet.Properties != null)
             {
-                propertiesWriter.WriteWillDelayInterval(packet.Properties.WillDelayInterval);
                 propertiesWriter.WriteSessionExpiryInterval(packet.Properties.SessionExpiryInterval);
                 propertiesWriter.WriteAuthenticationMethod(packet.Properties.AuthenticationMethod);
                 propertiesWriter.WriteAuthenticationData(packet.Properties.AuthenticationData);
                 propertiesWriter.WriteRequestProblemInformation(packet.Properties.RequestProblemInformation);
                 propertiesWriter.WriteRequestResponseInformation(packet.Properties.RequestResponseInformation);
                 propertiesWriter.WriteReceiveMaximum(packet.Properties.ReceiveMaximum);
-                propertiesWriter.WriteTopicAlias(packet.Properties.TopicAliasMaximum);
+                propertiesWriter.WriteTopicAliasMaximum(packet.Properties.TopicAliasMaximum);
                 propertiesWriter.WriteMaximumPacketSize(packet.Properties.MaximumPacketSize);
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             packetWriter.WriteWithLengthPrefix(packet.ClientId);
-
+            
             if (packet.WillMessage != null)
             {
+                var willPropertiesWriter = new MqttV500PropertiesWriter();
+                willPropertiesWriter.WritePayloadFormatIndicator(packet.WillMessage.PayloadFormatIndicator);
+                willPropertiesWriter.WriteMessageExpiryInterval(packet.WillMessage.MessageExpiryInterval);
+                willPropertiesWriter.WriteTopicAlias(packet.WillMessage.TopicAlias);
+                willPropertiesWriter.WriteResponseTopic(packet.WillMessage.ResponseTopic);
+                willPropertiesWriter.WriteCorrelationData(packet.WillMessage.CorrelationData);
+                willPropertiesWriter.WriteSubscriptionIdentifiers(packet.WillMessage.SubscriptionIdentifiers);
+                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.WillMessage.Payload);
             }
@@ -202,7 +216,7 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
         }
@@ -244,12 +258,12 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteTopicAlias(packet.Properties.TopicAlias);
                 propertiesWriter.WriteResponseTopic(packet.Properties.ResponseTopic);
                 propertiesWriter.WriteCorrelationData(packet.Properties.CorrelationData);
-                propertiesWriter.WriteSubscriptionIdentifier(packet.Properties.SubscriptionIdentifier);
+                propertiesWriter.WriteSubscriptionIdentifiers(packet.Properties.SubscriptionIdentifiers);
                 propertiesWriter.WriteContentType(packet.Properties.ContentType);
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             if (packet.Payload?.Length > 0)
             {
@@ -300,7 +314,7 @@ namespace MQTTnet.Formatter.V5
             if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubAckReasonCode.Success)
             {
                 packetWriter.Write((byte)packet.ReasonCode.Value);
-                propertiesWriter.WriteToPacket(packetWriter);
+                propertiesWriter.WriteTo(packetWriter);
             }
             
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
@@ -328,7 +342,7 @@ namespace MQTTnet.Formatter.V5
             if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRecReasonCode.Success)
             {
                 packetWriter.Write((byte)packet.ReasonCode.Value);
-                propertiesWriter.WriteToPacket(packetWriter);
+                propertiesWriter.WriteTo(packetWriter);
             }
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
@@ -343,9 +357,6 @@ namespace MQTTnet.Formatter.V5
                 ThrowReasonCodeNotSetException();
             }
 
-            packetWriter.Write(packet.PacketIdentifier.Value);
-            packetWriter.Write((byte)packet.ReasonCode.Value);
-
             var propertiesWriter = new MqttV500PropertiesWriter();
             if (packet.Properties != null)
             {
@@ -353,10 +364,12 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRelReasonCode.Success)
+            packetWriter.Write(packet.PacketIdentifier.Value);
+            
+            if (propertiesWriter.Length > 0 || packet.ReasonCode.Value != MqttPubRelReasonCode.Success)
             {
                 packetWriter.Write((byte)packet.ReasonCode.Value);
-                propertiesWriter.WriteToPacket(packetWriter);
+                propertiesWriter.WriteTo(packetWriter);
             }
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
@@ -372,8 +385,7 @@ namespace MQTTnet.Formatter.V5
             }
 
             packetWriter.Write(packet.PacketIdentifier.Value);
-            packetWriter.Write((byte)packet.ReasonCode.Value);
-
+            
             var propertiesWriter = new MqttV500PropertiesWriter();
             if (packet.Properties != null)
             {
@@ -381,10 +393,10 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            if (packetWriter.Length > 0 || packet.ReasonCode.Value != MqttPubCompReasonCode.Success)
+            if (propertiesWriter.Length > 0 || packet.ReasonCode.Value != MqttPubCompReasonCode.Success)
             {
                 packetWriter.Write((byte)packet.ReasonCode.Value);
-                propertiesWriter.WriteToPacket(packetWriter);
+                propertiesWriter.WriteTo(packetWriter);
             }
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
@@ -405,7 +417,7 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             if (packet.TopicFilters?.Count > 0)
             {
@@ -452,7 +464,7 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             foreach (var reasonCode in packet.ReasonCodes)
             {
@@ -476,7 +488,7 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             foreach (var topicFilter in packet.TopicFilters)
             {
@@ -501,7 +513,7 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             foreach (var reasonCode in packet.ReasonCodes)
             {
@@ -529,7 +541,7 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Disconnect);
         }
@@ -557,7 +569,7 @@ namespace MQTTnet.Formatter.V5
                 propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
             }
 
-            propertiesWriter.WriteToPacket(packetWriter);
+            propertiesWriter.WriteTo(packetWriter);
 
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Auth);
         }

Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using MQTTnet.Exceptions;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
@@ -43,24 +42,14 @@ namespace MQTTnet.Formatter.V5
             return true;
         }
 
-        public void FillUserProperties(List<MqttUserProperty> userProperties)
+        public void AddUserPropertyTo(List<MqttUserProperty> userProperties)
         {
             if (userProperties == null) throw new ArgumentNullException(nameof(userProperties));
 
-            var userPropertiesLength = _body.ReadVariableLengthInteger();
-            if (userPropertiesLength == 0)
-            {
-                return;
-            }
+            var name = _body.ReadStringWithLengthPrefix();
+            var value = _body.ReadStringWithLengthPrefix();
 
-            var targetPosition = _body.Offset + userPropertiesLength;
-            while (_body.Offset < targetPosition)
-            {
-                var name = _body.ReadStringWithLengthPrefix();
-                var value = _body.ReadStringWithLengthPrefix();
-
-                userProperties.Add(new MqttUserProperty(name, value));
-            }
+            userProperties.Add(new MqttUserProperty(name, value));
         }
 
         public string ReadReasonString()
@@ -78,17 +67,17 @@ namespace MQTTnet.Formatter.V5
             return _body.ReadWithLengthPrefix();
         }
 
-        public bool? ReadRetainAvailable()
+        public bool ReadRetainAvailable()
         {
             return _body.ReadBoolean();
         }
 
-        public uint? ReadSessionExpiryInterval()
+        public uint ReadSessionExpiryInterval()
         {
             return _body.ReadFourByteInteger();
         }
 
-        public ushort? ReadReceiveMaximum()
+        public ushort ReadReceiveMaximum()
         {
             return _body.ReadTwoByteInteger();
         }
@@ -103,17 +92,17 @@ namespace MQTTnet.Formatter.V5
             return _body.ReadStringWithLengthPrefix();
         }
 
-        public ushort? ReadTopicAliasMaximum()
+        public ushort ReadTopicAliasMaximum()
         {
             return _body.ReadTwoByteInteger();
         }
 
-        public uint? ReadMaximumPacketSize()
+        public uint ReadMaximumPacketSize()
         {
             return _body.ReadFourByteInteger();
         }
 
-        public ushort? ReadServerKeepAlive()
+        public ushort ReadServerKeepAlive()
         {
             return _body.ReadTwoByteInteger();
         }
@@ -123,22 +112,22 @@ namespace MQTTnet.Formatter.V5
             return _body.ReadStringWithLengthPrefix();
         }
 
-        public bool? ReadSharedSubscriptionAvailable()
+        public bool ReadSharedSubscriptionAvailable()
         {
             return _body.ReadBoolean();
         }
 
-        public bool? ReadSubscriptionIdentifiersAvailable()
+        public bool ReadSubscriptionIdentifiersAvailable()
         {
             return _body.ReadBoolean();
         }
 
-        public bool? ReadWildcardSubscriptionAvailable()
+        public bool ReadWildcardSubscriptionAvailable()
         {
             return _body.ReadBoolean();
         }
 
-        public uint? ReadSubscriptionIdentifier()
+        public uint ReadSubscriptionIdentifier()
         {
             return _body.ReadVariableLengthInteger();
         }
@@ -148,12 +137,12 @@ namespace MQTTnet.Formatter.V5
             return (MqttPayloadFormatIndicator)_body.ReadByte();
         }
 
-        public uint? ReadMessageExpiryInterval()
+        public uint ReadMessageExpiryInterval()
         {
             return _body.ReadFourByteInteger();
         }
 
-        public ushort? ReadTopicAlias()
+        public ushort ReadTopicAlias()
         {
             return _body.ReadTwoByteInteger();
         }
@@ -173,17 +162,17 @@ namespace MQTTnet.Formatter.V5
             return _body.ReadStringWithLengthPrefix();
         }
 
-        public uint? ReadWillDelayInterval()
+        public uint ReadWillDelayInterval()
         {
             return _body.ReadFourByteInteger();
         }
 
-        public bool? RequestResponseInformation()
+        public bool RequestResponseInformation()
         {
             return _body.ReadBoolean();
         }
 
-        public bool? RequestProblemInformation()
+        public bool RequestProblemInformation()
         {
             return _body.ReadBoolean();
         }

Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs

@@ -7,8 +7,11 @@ namespace MQTTnet.Formatter.V5
 {
     public class MqttV500PropertiesWriter
     {
+        // TODO: Consider lazy init on first write to avoid useless allocations.
         private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
 
+        public int Length => _packetWriter.Length;
+
         public void WriteUserProperties(List<MqttUserProperty> userProperties)
         {
             if (userProperties == null || userProperties.Count == 0)
@@ -16,16 +19,12 @@ namespace MQTTnet.Formatter.V5
                 return;
             }
 
-            var propertyWriter = new MqttPacketWriter();
             foreach (var property in userProperties)
             {
-                propertyWriter.WriteWithLengthPrefix(property.Name);
-                propertyWriter.WriteWithLengthPrefix(property.Value);
+                _packetWriter.Write((byte)MqttPropertyId.UserProperty);
+                _packetWriter.WriteWithLengthPrefix(property.Name);
+                _packetWriter.WriteWithLengthPrefix(property.Value);
             }
-
-            _packetWriter.Write((byte)MqttPropertyId.UserProperty);
-            _packetWriter.WriteVariableLengthInteger((uint)propertyWriter.Length);
-            _packetWriter.Write(propertyWriter);
         }
 
         public void WriteCorrelationData(byte[] value)
@@ -63,7 +62,7 @@ namespace MQTTnet.Formatter.V5
             Write(MqttPropertyId.AuthenticationMethod, value);
         }
 
-        public void WriteToPacket(IMqttPacketWriter packetWriter)
+        public void WriteTo(IMqttPacketWriter packetWriter)
         {
             if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
 
@@ -81,6 +80,19 @@ namespace MQTTnet.Formatter.V5
             WriteAsVariableLengthInteger(MqttPropertyId.SubscriptionIdentifier, value);
         }
 
+        public void WriteSubscriptionIdentifiers(IEnumerable<uint> value)
+        {
+            if (value == null)
+            {
+                return;
+            }
+
+            foreach (var subscriptionIdentifier in value)
+            {
+                WriteAsVariableLengthInteger(MqttPropertyId.SubscriptionIdentifier, subscriptionIdentifier);
+            }
+        }
+
         public void WriteTopicAlias(ushort? value)
         {
             Write(MqttPropertyId.TopicAlias, value);
@@ -118,7 +130,7 @@ namespace MQTTnet.Formatter.V5
 
         public void WriteReceiveMaximum(ushort? value)
         {
-            Write(MqttPropertyId.RequestResponseInformation, value);
+            Write(MqttPropertyId.ReceiveMaximum, value);
         }
 
         public void WriteMaximumPacketSize(uint? value)
@@ -133,7 +145,7 @@ namespace MQTTnet.Formatter.V5
 
         public void WriteAssignedClientIdentifier(string value)
         {
-            Write(MqttPropertyId.AssignedClientIdentifer, value);
+            Write(MqttPropertyId.AssignedClientIdentifier, value);
         }
 
         public void WriteTopicAliasMaximum(ushort? value)
@@ -177,6 +189,17 @@ namespace MQTTnet.Formatter.V5
             _packetWriter.Write(value.Value ? (byte)0x1 : (byte)0x0);
         }
 
+        private void Write(MqttPropertyId id, byte? value)
+        {
+            if (!value.HasValue)
+            {
+                return;
+            }
+
+            _packetWriter.Write((byte)id);
+            _packetWriter.Write(value.Value);
+        }
+
         private void Write(MqttPropertyId id, ushort? value)
         {
             if (!value.HasValue)

Source/MQTTnet/MqttApplicationMessage.cs

@@ -28,6 +28,6 @@ namespace MQTTnet
 
         public byte[] CorrelationData { get; set; }
 
-        public uint? SubscriptionIdentifier { get; set; }
+        public List<uint> SubscriptionIdentifiers { get; set; }
     }
 }

Source/MQTTnet/MqttApplicationMessageBuilder.cs

@@ -18,7 +18,7 @@ namespace MQTTnet
         private string _responseTopic;
         private byte[] _correlationData;
         private ushort? _topicAlias;
-        private uint? _subscriptionIdentifier;
+        private List<uint> _subscriptionIdentifiers;
         private uint? _messageExpiryInterval;
         private MqttPayloadFormatIndicator? _payloadFormatIndicator;
         private List<MqttUserProperty> _userProperties;
@@ -43,7 +43,7 @@ namespace MQTTnet
             {
                 _payload = payload.ToArray();
             }
-            
+
             return this;
         }
 
@@ -176,7 +176,12 @@ namespace MQTTnet
         /// </summary>
         public MqttApplicationMessageBuilder WithSubscriptionIdentifier(uint subscriptionIdentifier)
         {
-            _subscriptionIdentifier = subscriptionIdentifier;
+            if (_subscriptionIdentifiers == null)
+            {
+                _subscriptionIdentifiers = new List<uint>();
+            }
+
+            _subscriptionIdentifiers.Add(subscriptionIdentifier);
             return this;
         }
 
@@ -215,16 +220,12 @@ namespace MQTTnet
                 ResponseTopic = _responseTopic,
                 CorrelationData = _correlationData,
                 TopicAlias = _topicAlias,
-                SubscriptionIdentifier = _subscriptionIdentifier,
+                SubscriptionIdentifiers = _subscriptionIdentifiers,
                 MessageExpiryInterval = _messageExpiryInterval,
-                PayloadFormatIndicator = _payloadFormatIndicator
+                PayloadFormatIndicator = _payloadFormatIndicator,
+                UserProperties = _userProperties
             };
 
-            if (_userProperties?.Any() == true)
-            {
-                applicationMessage.UserProperties = _userProperties;
-            }
-            
             return applicationMessage;
         }
     }

Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs

@@ -23,6 +23,16 @@ namespace MQTTnet.PacketDispatcher
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
+            if (packet is MqttDisconnectPacket disconnectPacket)
+            {
+                foreach (var packetAwaiter in _packetAwaiters)
+                {
+                    packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket));
+                }
+
+                return;
+            }
+
             ushort identifier = 0;
             if (packet is IMqttPacketWithIdentifier packetWithIdentifier && packetWithIdentifier.PacketIdentifier.HasValue)
             {

Source/MQTTnet/Packets/MqttPublishPacketProperties.cs

@@ -15,9 +15,9 @@ namespace MQTTnet.Packets
 
         public byte[] CorrelationData { get; set; }
 
-        public List<MqttUserProperty> UserProperties { get; } = new List<MqttUserProperty>();
+        public List<MqttUserProperty> UserProperties { get; set; }
 
-        public uint? SubscriptionIdentifier { get; set; }
+        public List<uint> SubscriptionIdentifiers { get; set; }
 
         public string ContentType { get; set; }
     }

Source/MQTTnet/Protocol/MqttPropertyID.cs

@@ -9,7 +9,7 @@
         CorrelationData = 9,
         SubscriptionIdentifier = 11,
         SessionExpiryInterval = 17,
-        AssignedClientIdentifer = 18,
+        AssignedClientIdentifier = 18,
         ServerKeepAlive = 19,
         AuthenticationMethod = 21,
         AuthenticationData = 22,

Source/MQTTnet/Server/MqttClientConnection.cs

@@ -138,7 +138,7 @@ namespace MQTTnet.Server
                     {
                         ReturnCode = MqttConnectReturnCode.ConnectionAccepted,
                         ReasonCode = MqttConnectReasonCode.Success,
-                        IsSessionPresent = Session.IsCleanSession
+                        IsSessionPresent = !Session.IsCleanSession
                     }).ConfigureAwait(false);
 
                 Session.IsCleanSession = false;

Source/MQTTnet/Server/MqttRetainedMessagesManager.cs

@@ -61,8 +61,9 @@ namespace MQTTnet.Server
                 using (await _messagesLock.WaitAsync().ConfigureAwait(false))
                 {
                     var saveIsRequired = false;
+                    var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
 
-                    if (applicationMessage.Payload?.Any() != true)
+                    if (!hasPayload)
                     {
                         saveIsRequired = _messages.Remove(applicationMessage.Topic);
                         _logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);

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

@@ -10,12 +10,47 @@ using MQTTnet.Client.Unsubscribing;
 using MQTTnet.Formatter;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
+using MQTTnet.Tests.Mockups;
 
 namespace MQTTnet.Tests.MQTTv5
 {
     [TestClass]
     public class Client_Tests
     {
+        [TestMethod]
+        public async Task Connect_With_New_Mqtt_Features()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+
+                // This test can be also executed against "broker.hivemq.com" to validate package format.
+                var client = await testEnvironment.ConnectClientAsync(
+                    new MqttClientOptionsBuilder()
+                        //.WithTcpServer("broker.hivemq.com")
+                        .WithTcpServer("127.0.0.1", testEnvironment.ServerPort)
+                        .WithProtocolVersion(MqttProtocolVersion.V500)
+                        .WithTopicAliasMaximum(20)
+                        .WithReceiveMaximum(20)
+                        .WithWillMessage(new MqttApplicationMessageBuilder().WithTopic("abc").Build())
+                        .WithWillDelayInterval(20)
+                        .Build());
+
+                await client.SubscribeAsync("a");
+
+                await client.PublishAsync(new MqttApplicationMessageBuilder()
+                    .WithTopic("a")
+                    .WithPayload("x")
+                    .WithUserProperty("a", "1")
+                    .WithUserProperty("b", "2")
+                    .WithPayloadFormatIndicator(MqttPayloadFormatIndicator.CharacterData)
+                    .WithAtLeastOnceQoS()
+                    .Build());
+
+                await Task.Delay(500);
+            }
+        }
+
         [TestMethod]
         public async Task Connect()
         {
@@ -63,7 +98,16 @@ namespace MQTTnet.Tests.MQTTv5
                 await server.StartAsync(new MqttServerOptions());
 
                 await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1").WithProtocolVersion(MqttProtocolVersion.V500).Build());
-                var result = await client.SubscribeAsync("a", MqttQualityOfServiceLevel.AtLeastOnce);
+
+                var result = await client.SubscribeAsync(new MqttClientSubscribeOptions()
+                {
+                    SubscriptionIdentifier = 1,
+                    TopicFilters = new List<TopicFilter>
+                    {
+                        new TopicFilter { Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}
+                    }
+                });
+
                 await client.DisconnectAsync();
 
                 Assert.AreEqual(1, result.Items.Count);

Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs

@@ -0,0 +1,40 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Client;
+using MQTTnet.Client.Options;
+using MQTTnet.Formatter;
+using MQTTnet.Tests.Mockups;
+
+namespace MQTTnet.Tests.MQTTv5
+{
+    [TestClass]
+    public class Server_Tests
+    {
+        [TestMethod]
+        public async Task Will_Message_Send()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                var receivedMessagesCount = 0;
+
+                await testEnvironment.StartServerAsync();
+
+                var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
+
+                var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage).WithProtocolVersion(MqttProtocolVersion.V500);
+
+                var c1 = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
+                c1.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
+                await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build());
+
+                var c2 = await testEnvironment.ConnectClientAsync(clientOptions);
+                c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
+
+                await Task.Delay(1000);
+
+                Assert.AreEqual(1, receivedMessagesCount);
+            }
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs

@@ -90,12 +90,24 @@ namespace MQTTnet.Tests.Mockups
 
         public async Task<IMqttClient> ConnectClientAsync(MqttClientOptionsBuilder options)
         {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
             var client = CreateClient();
             await client.ConnectAsync(options.WithTcpServer("localhost", ServerPort).Build());
 
             return client;
         }
 
+        public async Task<IMqttClient> ConnectClientAsync(IMqttClientOptions options)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+
+            var client = CreateClient();
+            await client.ConnectAsync(options);
+
+            return client;
+        }
+
         public void ThrowIfLogErrors()
         {
             lock (_serverErrors)

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -11,7 +11,6 @@ using MQTTnet.Client.Disconnecting;
 using MQTTnet.Client.Options;
 using MQTTnet.Client.Subscribing;
 using MQTTnet.Exceptions;
-using MQTTnet.Formatter;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 using MQTTnet.Tests.Mockups;
@@ -21,6 +20,34 @@ namespace MQTTnet.Tests
     [TestClass]
     public class Client_Tests
     {
+        [TestMethod]
+        public async Task PacketIdentifier_In_Publish_Result()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+                var client = await testEnvironment.ConnectClientAsync();
+
+                var result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.AtMostOnce);
+                Assert.AreEqual(null, result.PacketIdentifier);
+
+                result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.AtMostOnce);
+                Assert.AreEqual(null, result.PacketIdentifier);
+
+                result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.AtLeastOnce);
+                Assert.AreEqual((ushort)1, result.PacketIdentifier);
+
+                result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.AtLeastOnce);
+                Assert.AreEqual((ushort)2, result.PacketIdentifier);
+
+                result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.ExactlyOnce);
+                Assert.AreEqual((ushort)3, result.PacketIdentifier);
+
+                result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.ExactlyOnce);
+                Assert.AreEqual((ushort)4, result.PacketIdentifier);
+            }
+        }
+        
         [TestMethod]
         public async Task Invalid_Connect_Throws_Exception()
         {

Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.IO;
 using System.Linq;
 using System.Text;
@@ -234,7 +235,7 @@ namespace MQTTnet.Tests
         [TestMethod]
         public void SerializeV500_MqttPublishPacket()
         {
-            var prop = new MqttPublishPacketProperties();
+            var prop = new MqttPublishPacketProperties {UserProperties = new List<MqttUserProperty>()};
 
             prop.ResponseTopic = "/Response";
 

Tests/MQTTnet.Core.Tests/MqttTcpChannel_Tests.cs

@@ -15,11 +15,11 @@ namespace MQTTnet.Tests
         public async Task Dispose_Channel_While_Used()
         {
             var ct = new CancellationTokenSource();
+            var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
 
             try
             {
-                var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
-                serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50000));
+                serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
                 serverSocket.Listen(0);
 
 #pragma warning disable 4014
@@ -35,7 +35,7 @@ namespace MQTTnet.Tests
                 }, ct.Token);
 
                 var clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
-                await clientSocket.ConnectAsync(IPAddress.Loopback, 50000);
+                await clientSocket.ConnectAsync(IPAddress.Loopback, 50001);
 
                 await Task.Delay(100, ct.Token);
 
@@ -68,6 +68,7 @@ namespace MQTTnet.Tests
             finally
             {
                 ct.Cancel(false);
+                serverSocket.Dispose();
             }
         }
     }

Tests/MQTTnet.Core.Tests/Server_Tests.cs

@@ -54,6 +54,20 @@ namespace MQTTnet.Tests
                 1);
         }
 
+        [TestMethod]
+        public async Task Use_Clean_Session()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                await testEnvironment.StartServerAsync();
+
+                var client = testEnvironment.CreateClient();
+                var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build());
+
+                Assert.IsFalse(connectResult.IsSessionPresent);
+            }
+        }
+
         [TestMethod]
         public async Task Will_Message_Do_Not_Send()
         {
@@ -61,7 +75,7 @@ namespace MQTTnet.Tests
             {
                 var receivedMessagesCount = 0;
 
-                await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder());
+                await testEnvironment.StartServerAsync();
 
                 var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
 
@@ -540,7 +554,7 @@ namespace MQTTnet.Tests
         }
 
         [TestMethod]
-        public async Task Clear_Retained_Message()
+        public async Task Clear_Retained_Message_With_Empty_Payload()
         {
             using (var testEnvironment = new TestEnvironment())
             {
@@ -566,7 +580,35 @@ namespace MQTTnet.Tests
                 Assert.AreEqual(0, receivedMessagesCount);
             }
         }
-        
+
+        [TestMethod]
+        public async Task Clear_Retained_Message_With_Null_Payload()
+        {
+            using (var testEnvironment = new TestEnvironment())
+            {
+                var receivedMessagesCount = 0;
+
+                await testEnvironment.StartServerAsync();
+
+                var c1 = await testEnvironment.ConnectClientAsync();
+
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
+                await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
+
+                await c1.DisconnectAsync();
+
+                var c2 = await testEnvironment.ConnectClientAsync();
+
+                c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
+
+                await Task.Delay(200);
+                await c2.SubscribeAsync(new TopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
+                await Task.Delay(500);
+
+                Assert.AreEqual(0, receivedMessagesCount);
+            }
+        }
+
         [TestMethod]
         public async Task Intercept_Application_Message()
         {

Tests/MQTTnet.TestApp.NetCore/Program.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Net.Security;
+using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Client;
@@ -81,7 +83,7 @@ namespace MQTTnet.TestApp.NetCore
             {
                 Task.Run(PerformanceTest.RunQoS0Test);
             }
-            
+
             Thread.Sleep(Timeout.Infinite);
         }
     }
@@ -118,4 +120,22 @@ namespace MQTTnet.TestApp.NetCore
             return Task.FromResult(retainedMessages);
         }
     }
+
+    public class WikiCode
+    {
+        public void Code()
+        {
+            //Validate certificate.
+            var options = new MqttClientOptionsBuilder()
+                .WithTls(new MqttClientOptionsBuilderTlsParameters
+                {
+                    CertificateValidationCallback = (X509Certificate x, X509Chain y, SslPolicyErrors z, IMqttClientOptions o) =>
+                        {
+                            // TODO: Check conditions of certificate by using above parameters.
+                            return true;
+                        }
+                })
+                .Build();
+        }
+    }
 }

Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.ObjectModel;
+using System.Net.Security;
+using System.Security.Cryptography.X509Certificates;
 using System.Text;
 using System.Threading.Tasks;
 using Windows.Security.Cryptography.Certificates;
@@ -595,7 +597,7 @@ namespace MQTTnet.TestApp.UniversalWindows
                     };
                 }
             }
-
+            
             // ----------------------------------
             {
                 var options = new MqttServerOptions();
@@ -714,6 +716,25 @@ namespace MQTTnet.TestApp.UniversalWindows
                 var mqttServer = new MqttFactory().CreateMqttServer();
             }
 
+            {
+                // Setup subscription interceptor.
+                var options = new MqttServerOptionsBuilder()
+                    .WithSubscriptionInterceptor(context =>
+                    {
+                        if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
+                        {
+                            context.AcceptSubscription = false;
+                        }
+
+                        if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
+                        {
+                            context.AcceptSubscription = false;
+                            context.CloseConnection = true;
+                        }
+                    })
+                    .Build();
+            }
+
             {
                 // Setup and start a managed MQTT client.
                 var options = new ManagedMqttClientOptionsBuilder()

_config.yml

@@ -1 +1 @@
-theme: jekyll-theme-modernist
+theme: jekyll-theme-hacker

appveyor.yml

@@ -14,5 +14,5 @@ build:
   verbosity: minimal
 
 test_script:
-- cmd: dotnet vstest "%APPVEYOR_BUILD_FOLDER%\Tests\MQTTnet.Core.Tests\bin\Release\netcoreapp2.1\MQTTnet.Core.Tests.dll"
+- cmd: dotnet vstest "%APPVEYOR_BUILD_FOLDER%\Tests\MQTTnet.Core.Tests\bin\Release\netcoreapp2.1\MQTTnet.Tests.dll"
 - cmd: dotnet vstest "%APPVEYOR_BUILD_FOLDER%\Tests\MQTTnet.AspNetCore.Tests\bin\Release\netcoreapp2.1\MQTTnet.AspNetCore.Tests.dll"