Refactor property definition for packets.

Source/MQTTnet/Formatter/IMqttPacketFormatter.cs

@@ -10,6 +10,8 @@ namespace MQTTnet.Formatter
 
         MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket);
 
+        MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage);
+
         void FreeBuffer();
     }
 }

Source/MQTTnet/Formatter/MqttFixedHeader.cs

@@ -2,7 +2,7 @@
 {
     public struct MqttFixedHeader
     {
-        public MqttFixedHeader(byte flags, int remainingLength)
+        public MqttFixedHeader(byte flags, uint remainingLength)
         {
             Flags = flags;
             RemainingLength = remainingLength;
@@ -10,6 +10,6 @@
 
         public byte Flags { get; }
 
-        public int RemainingLength { get; }
+        public uint RemainingLength { get; }
     }
 }

Source/MQTTnet/Formatter/MqttPacketBodyReader.cs

@@ -1,71 +1,140 @@
 using System;
+using System.Runtime.CompilerServices;
 using System.Text;
+using MQTTnet.Exceptions;
 
 namespace MQTTnet.Formatter
 {
     public class MqttPacketBodyReader
     {
         private readonly byte[] _buffer;
-        private readonly int _initialOffset;
-        private readonly int _length;
+        private readonly ulong _initialOffset;
+        private readonly ulong _length;
 
-        private int _offset;
-        
         public MqttPacketBodyReader(byte[] buffer, int offset, int length)
+         : this(buffer, (ulong)offset, (ulong)length)
+        {
+        }
+
+        public MqttPacketBodyReader(byte[] buffer, ulong offset, ulong length)
         {
             _buffer = buffer;
             _initialOffset = offset;
-            _offset = offset;
+            Offset = offset;
             _length = length;
         }
 
-        public int Length => _length - _offset;
+        public ulong Offset { get; private set; }
 
-        public bool EndOfStream => _offset == _length;
+        public ulong Length => _length - Offset;
 
-        public void Seek(int position)
+        public bool EndOfStream => Offset == _length;
+
+        public void Seek(ulong position)
         {
-            _offset = _initialOffset + position;
+            Offset = _initialOffset + position;
+        }
+
+        public ArraySegment<byte> Read(uint length)
+        {
+            ValidateReceiveBuffer(length);
+
+            var buffer = new ArraySegment<byte>(_buffer, (int)Offset, (int)length);
+            Offset += length;
+            return buffer;
         }
 
         public byte ReadByte()
         {
             ValidateReceiveBuffer(1);
-            return _buffer[_offset++];
+            return _buffer[Offset++];
+        }
+
+        public bool ReadBoolean()
+        {
+            ValidateReceiveBuffer(1);
+            var buffer = _buffer[Offset++];
+
+            if (buffer == 0)
+            {
+                return false;
+            }
+
+            if (buffer == 1)
+            {
+                return true;
+            }
+
+            throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only.");
         }
 
         public ArraySegment<byte> ReadRemainingData()
         {
-            return new ArraySegment<byte>(_buffer, _offset, _length - _offset);
+            return new ArraySegment<byte>(_buffer, (int)Offset, (int)(_length - Offset));
         }
 
-        public ushort ReadUInt16()
+        public ushort ReadTwoByteInteger()
         {
             ValidateReceiveBuffer(2);
 
-            var msb = _buffer[_offset++];
-            var lsb = _buffer[_offset++];
+            var msb = _buffer[Offset++];
+            var lsb = _buffer[Offset++];
             
             return (ushort)(msb << 8 | lsb);
         }
 
+        public uint ReadFourByteInteger()
+        {
+            ValidateReceiveBuffer(4);
+
+            var byte0 = _buffer[Offset++];
+            var byte1 = _buffer[Offset++];
+            var byte2 = _buffer[Offset++];
+            var byte3 = _buffer[Offset++];
+
+            return (uint)(byte0 << 24 | byte1 << 16 | byte2 << 8 | byte3);
+        }
+
         public ArraySegment<byte> ReadWithLengthPrefix()
         {
-            var length = ReadUInt16();
+            var length = ReadTwoByteInteger();
 
             ValidateReceiveBuffer(length);
 
-            var result = new ArraySegment<byte>(_buffer, _offset, length);
-            _offset += length;
+            var result = new ArraySegment<byte>(_buffer, (int)Offset, length);
+            Offset += length;
 
             return result;
         }
 
-        private void ValidateReceiveBuffer(ushort length)
+        public uint ReadVariableLengthInteger()
         {
-            if (_length < _offset + length)
+            var multiplier = 1;
+            var value = 0U;
+            byte encodedByte;
+
+            do
             {
-                throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_length} bytes");
+                encodedByte = ReadByte();
+                value += (uint)((encodedByte & 127) * multiplier);
+
+                if (multiplier > 2097152)
+                {
+                    throw new MqttProtocolViolationException("Variable length integer is invalid.");
+                }
+
+                multiplier *= 128;
+            } while ((encodedByte & 128) != 0);
+
+            return value;
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void ValidateReceiveBuffer(uint length)
+        {
+            if (_length < Offset + length)
+            {
+                throw new ArgumentOutOfRangeException(nameof(_buffer), $"Expected at least {Offset + length} bytes but there are only {_length} bytes");
             }
         }
 

Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs

@@ -8,44 +8,56 @@ using MQTTnet.Packets;
 
 namespace MQTTnet.Formatter
 {
-    public class MqttPacketSerializerAdapter
+    public class MqttPacketFormatterAdapter
     {
-        public MqttPacketSerializerAdapter()
+        private IMqttPacketFormatter _formatter;
+
+        public MqttPacketFormatterAdapter()
         {
         }
 
-        public MqttPacketSerializerAdapter(MqttProtocolVersion protocolVersion)
+        public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion)
         {
             UseProtocolVersion(protocolVersion);
         }
 
         public MqttProtocolVersion? ProtocolVersion { get; private set; }
-
-        public IMqttPacketFormatter Formatter { get; private set; }
-
-        public ArraySegment<byte> Encode(MqttBasePacket packet)
+        
+        public MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage)
         {
-            if (Formatter == null)
+            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+
+            if (_formatter == null)
             {
                 throw new InvalidOperationException("Protocol version not set or detected.");
             }
 
-            return Formatter.Encode(packet);
+            return _formatter.ConvertApplicationMessageToPublishPacket(applicationMessage);
+        }
+
+        public ArraySegment<byte> Encode(MqttBasePacket packet)
+        {
+            if (_formatter == null)
+            {
+                throw new InvalidOperationException("Protocol version not set or detected.");
+            }
+
+            return _formatter.Encode(packet);
         }
 
         public MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket)
         {
-            if (Formatter == null)
+            if (_formatter == null)
             {
                 throw new InvalidOperationException("Protocol version not set or detected.");
             }
 
-            return Formatter.Decode(receivedMqttPacket);
+            return _formatter.Decode(receivedMqttPacket);
         }
 
         public void FreeBuffer()
         {
-            Formatter?.FreeBuffer();
+            _formatter?.FreeBuffer();
         }
 
         public void DetectProtocolVersion(ReceivedMqttPacket receivedMqttPacket)
@@ -68,20 +80,20 @@ namespace MQTTnet.Formatter
             {
                 case MqttProtocolVersion.V500:
                     {
-                        Formatter = new MqttV500PacketFormatter();
+                        _formatter = new MqttV500PacketFormatter();
                         break;
                     }
 
                 case MqttProtocolVersion.V311:
                     {
-                        Formatter = new MqttV311PacketFormatter();
+                        _formatter = new MqttV311PacketFormatter();
                         break;
                     }
 
                 case MqttProtocolVersion.V310:
                     {
 
-                        Formatter = new MqttV310PacketFormatter();
+                        _formatter = new MqttV310PacketFormatter();
                         break;
                     }
                     

Source/MQTTnet/Formatter/MqttPacketReader.cs

@@ -1,4 +1,5 @@
-using System.Threading;
+using System;
+using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Channel;
 using MQTTnet.Exceptions;
@@ -6,9 +7,18 @@ using MQTTnet.Internal;
 
 namespace MQTTnet.Formatter
 {
-    public static class MqttPacketReader
+    public class MqttPacketReader
     {
-        public static async Task<MqttFixedHeader> ReadFixedHeaderAsync(IMqttChannel channel, byte[] fixedHeaderBuffer, byte[] singleByteBuffer, CancellationToken cancellationToken)
+        private readonly byte[] _singleByteBuffer = new byte[1];
+
+        private readonly IMqttChannel _channel;
+
+        public MqttPacketReader(IMqttChannel channel)
+        {
+            _channel = channel ?? throw new ArgumentNullException(nameof(channel));
+        }
+
+        public async Task<MqttFixedHeader> ReadFixedHeaderAsync(byte[] fixedHeaderBuffer, CancellationToken cancellationToken)
         {
             // The MQTT fixed header contains 1 byte of flags and at least 1 byte for the remaining data length.
             // So in all cases at least 2 bytes must be read for a complete MQTT packet.
@@ -19,7 +29,7 @@ namespace MQTTnet.Formatter
 
             while (totalBytesRead < buffer.Length)
             {
-                var bytesRead = await channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
+                var bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false);
 
                 cancellationToken.ThrowIfCancellationRequested();
                 ExceptionHelper.ThrowIfGracefulSocketClose(bytesRead);
@@ -35,24 +45,24 @@ namespace MQTTnet.Formatter
 
 #if WINDOWS_UWP
             // UWP will have a dead lock when calling this not async.
-            var bodyLength = await ReadBodyLengthAsync(channel, buffer[1], singleByteBuffer, cancellationToken).ConfigureAwait(false);
+            var bodyLength = await ReadBodyLengthAsync(buffer[1], cancellationToken).ConfigureAwait(false);
 #else
             // Here the async/await pattern is not used becuase the overhead of context switches
             // is too big for reading 1 byte in a row. We expect that the remaining data was sent
             // directly after the initial bytes. If the client disconnects just in this moment we
             // will get an exception anyway.
-            var bodyLength = ReadBodyLength(channel, buffer[1], singleByteBuffer, cancellationToken);
+            var bodyLength = ReadBodyLength(buffer[1], cancellationToken);
 #endif
 
             return new MqttFixedHeader(buffer[0], bodyLength);
         }
 
 #if !WINDOWS_UWP
-        private static int ReadBodyLength(IMqttChannel channel, byte initialEncodedByte, byte[] singleByteBuffer, CancellationToken cancellationToken)
+        private uint ReadBodyLength(byte initialEncodedByte, CancellationToken cancellationToken)
         {
             var offset = 0;
             var multiplier = 128;
-            var value = initialEncodedByte & 127;
+            var value = (uint)(initialEncodedByte & 127);
             int encodedByte = initialEncodedByte;
 
             while ((encodedByte & 128) != 0)
@@ -65,18 +75,18 @@ namespace MQTTnet.Formatter
 
                 cancellationToken.ThrowIfCancellationRequested();
 
-                encodedByte = ReadByte(channel, singleByteBuffer, cancellationToken);
+                encodedByte = ReadByte(cancellationToken);
 
-                value += (byte)(encodedByte & 127) * multiplier;
+                value += (uint)((encodedByte & 127) * multiplier);
                 multiplier *= 128;
             }
 
             return value;
         }
 
-        private static byte ReadByte(IMqttChannel channel, byte[] singleByteBuffer, CancellationToken cancellationToken)
+        private byte ReadByte(CancellationToken cancellationToken)
         {
-            var readCount = channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
+            var readCount = _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
 
             cancellationToken.ThrowIfCancellationRequested();
 
@@ -85,16 +95,16 @@ namespace MQTTnet.Formatter
                 ExceptionHelper.ThrowGracefulSocketClose();
             }
 
-            return singleByteBuffer[0];
+            return _singleByteBuffer[0];
         }
 
 #else
         
-        private static async Task<int> ReadBodyLengthAsync(IMqttChannel channel, byte initialEncodedByte, byte[] singleByteBuffer, CancellationToken cancellationToken)
+        private async Task<uint> ReadBodyLengthAsync(byte initialEncodedByte, CancellationToken cancellationToken)
         {
             var offset = 0;
             var multiplier = 128;
-            var value = initialEncodedByte & 127;
+            var value = (uint)(initialEncodedByte & 127);
             int encodedByte = initialEncodedByte;
 
             while ((encodedByte & 128) != 0)
@@ -107,18 +117,18 @@ namespace MQTTnet.Formatter
 
                 cancellationToken.ThrowIfCancellationRequested();
 
-                encodedByte = await ReadByteAsync(channel, singleByteBuffer, cancellationToken).ConfigureAwait(false);
+                encodedByte = await ReadByteAsync(cancellationToken).ConfigureAwait(false);
 
-                value += (byte)(encodedByte & 127) * multiplier;
+                value += (uint)((encodedByte & 127) * multiplier);
                 multiplier *= 128;
             }
 
             return value;
         }
 
-        private static async Task<byte> ReadByteAsync(IMqttChannel channel, byte[] singleByteBuffer, CancellationToken cancellationToken)
+        private async Task<byte> ReadByteAsync(CancellationToken cancellationToken)
         {
-            var readCount = await channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false);
+            var readCount = await _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false);
 
             cancellationToken.ThrowIfCancellationRequested();
 
@@ -127,7 +137,7 @@ namespace MQTTnet.Formatter
                 ExceptionHelper.ThrowGracefulSocketClose();
             }
 
-            return singleByteBuffer[0];
+            return _singleByteBuffer[0];
         }
 
 #endif

Source/MQTTnet/Formatter/MqttPacketWriter.cs

@@ -17,7 +17,7 @@ namespace MQTTnet.Formatter
 
         private byte[] _buffer = new byte[128];
 
-        private int _position;
+        private int _offset;
 
         public int Length { get; private set; }
 
@@ -77,7 +77,7 @@ namespace MQTTnet.Formatter
         {
             EnsureAdditionalCapacity(1);
 
-            _buffer[_position] = @byte;
+            _buffer[_offset] = @byte;
             IncreasePostition(1);
         }
 
@@ -85,9 +85,9 @@ namespace MQTTnet.Formatter
         {
             EnsureAdditionalCapacity(2);
 
-            _buffer[_position] = (byte)(value >> 8);
+            _buffer[_offset] = (byte)(value >> 8);
             IncreasePostition(1);
-            _buffer[_position] = (byte)value;
+            _buffer[_offset] = (byte)value;
             IncreasePostition(1);
         }
 
@@ -97,7 +97,7 @@ namespace MQTTnet.Formatter
 
             EnsureAdditionalCapacity(count);
 
-            Array.Copy(buffer, offset, _buffer, _position, count);
+            Array.Copy(buffer, offset, _buffer, _offset, count);
             IncreasePostition(count);
         }
 
@@ -111,7 +111,12 @@ namespace MQTTnet.Formatter
         public void Write(MqttPacketWriter propertyWriter)
         {
             if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));
-            
+
+            if (propertyWriter.Length == 0)
+            {
+                return;
+            }
+
             Write(propertyWriter._buffer, 0, propertyWriter.Length);
         }
 
@@ -123,7 +128,7 @@ namespace MQTTnet.Formatter
         public void Seek(int position)
         {
             EnsureCapacity(position);
-            _position = position;
+            _offset = position;
         }
 
         public byte[] GetBuffer()
@@ -150,7 +155,7 @@ namespace MQTTnet.Formatter
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void EnsureAdditionalCapacity(int additionalCapacity)
         {
-            var freeSpace = _buffer.Length - _position;
+            var freeSpace = _buffer.Length - _offset;
             if (freeSpace >= additionalCapacity)
             {
                 return;
@@ -180,11 +185,11 @@ namespace MQTTnet.Formatter
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void IncreasePostition(int length)
         {
-            _position += length;
+            _offset += length;
 
-            if (_position > Length)
+            if (_offset > Length)
             {
-                Length = _position;
+                Length = _offset;
             }
         }
     }

Source/MQTTnet/Formatter/V310/MqttV310PacketFormatter.cs

@@ -70,6 +70,23 @@ namespace MQTTnet.Formatter.V310
             }
         }
 
+        public virtual MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage.UserProperties != null)
+            {
+                throw new MqttProtocolViolationException("User properties are not supported in MQTT version 3.");
+            }
+
+            return new MqttPublishPacket
+            {
+                Topic = applicationMessage.Topic,
+                Payload = applicationMessage.Payload,
+                QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel,
+                Retain = applicationMessage.Retain,
+                Dup = false
+            };
+        }
+
         public void FreeBuffer()
         {
             _packetWriter.FreeBuffer();
@@ -103,7 +120,7 @@ namespace MQTTnet.Formatter.V310
 
             return new MqttUnsubAckPacket
             {
-                PacketIdentifier = body.ReadUInt16()
+                PacketIdentifier = body.ReadTwoByteInteger()
             };
         }
 
@@ -113,7 +130,7 @@ namespace MQTTnet.Formatter.V310
 
             return new MqttPubCompPacket
             {
-                PacketIdentifier = body.ReadUInt16()
+                PacketIdentifier = body.ReadTwoByteInteger()
             };
         }
 
@@ -123,7 +140,7 @@ namespace MQTTnet.Formatter.V310
 
             return new MqttPubRelPacket
             {
-                PacketIdentifier = body.ReadUInt16()
+                PacketIdentifier = body.ReadTwoByteInteger()
             };
         }
 
@@ -133,7 +150,7 @@ namespace MQTTnet.Formatter.V310
 
             return new MqttPubRecPacket
             {
-                PacketIdentifier = body.ReadUInt16()
+                PacketIdentifier = body.ReadTwoByteInteger()
             };
         }
 
@@ -143,7 +160,7 @@ namespace MQTTnet.Formatter.V310
 
             return new MqttPubAckPacket
             {
-                PacketIdentifier = body.ReadUInt16()
+                PacketIdentifier = body.ReadTwoByteInteger()
             };
         }
 
@@ -153,7 +170,7 @@ namespace MQTTnet.Formatter.V310
 
             var packet = new MqttUnsubscribePacket
             {
-                PacketIdentifier = body.ReadUInt16(),
+                PacketIdentifier = body.ReadTwoByteInteger(),
             };
 
             while (!body.EndOfStream)
@@ -170,7 +187,7 @@ namespace MQTTnet.Formatter.V310
 
             var packet = new MqttSubscribePacket
             {
-                PacketIdentifier = body.ReadUInt16()
+                PacketIdentifier = body.ReadTwoByteInteger()
             };
 
             while (!body.EndOfStream)
@@ -196,7 +213,7 @@ namespace MQTTnet.Formatter.V310
             ushort? packetIdentifier = null;
             if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
             {
-                packetIdentifier = receivedMqttPacket.Body.ReadUInt16();
+                packetIdentifier = receivedMqttPacket.Body.ReadTwoByteInteger();
             }
 
             var packet = new MqttPublishPacket
@@ -236,7 +253,7 @@ namespace MQTTnet.Formatter.V310
             var passwordFlag = (connectFlags & 0x40) > 0;
             var usernameFlag = (connectFlags & 0x80) > 0;
 
-            packet.KeepAlivePeriod = body.ReadUInt16();
+            packet.KeepAlivePeriod = body.ReadTwoByteInteger();
             packet.ClientId = body.ReadStringWithLengthPrefix();
 
             if (willFlag)
@@ -270,7 +287,7 @@ namespace MQTTnet.Formatter.V310
 
             var packet = new MqttSubAckPacket
             {
-                PacketIdentifier = body.ReadUInt16()
+                PacketIdentifier = body.ReadTwoByteInteger()
             };
 
             while (!body.EndOfStream)
@@ -394,7 +411,7 @@ namespace MQTTnet.Formatter.V310
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
         }
 
-        private static byte SerializePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
+        protected virtual byte SerializePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
         {
             ValidatePublishPacket(packet);
 
@@ -439,7 +456,7 @@ namespace MQTTnet.Formatter.V310
             return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
         }
 
-        private static byte SerializePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
+        protected virtual byte SerializePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
         {
             if (!packet.PacketIdentifier.HasValue)
             {

Source/MQTTnet/Formatter/V500/MqttV500PacketDecoder.cs

@@ -1,6 +1,135 @@
-namespace MQTTnet.Formatter.V500
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using MQTTnet.Exceptions;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Formatter.V500
 {
     public class MqttV500PacketDecoder
     {
+        public MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
+        {
+            if (body == null) throw new ArgumentNullException(nameof(body));
+
+            ThrowIfBodyIsEmpty(body);
+
+            var packet = new MqttConnAckPacket();
+
+            var acknowledgeFlags = body.ReadByte();
+
+            packet.IsSessionPresent = (acknowledgeFlags & 0x1) > 0;
+            packet.ConnectReturnCode = (MqttConnectReturnCode)body.ReadByte();
+            packet.ConnectReasonCode = (MqttConnectReasonCode)body.ReadByte();
+
+            var propertiesLength = body.ReadVariableLengthInteger();
+            if (propertiesLength > 0)
+            {
+                packet.Properties = new MqttConnAckPacketProperties();
+
+                while (!body.EndOfStream)
+                {
+                    var propertyID = (MqttPropertyID)body.ReadByte();
+
+                    if (propertyID == MqttPropertyID.SessionExpiryInterval)
+                    {
+                        packet.Properties.SessionExpiryInterval = body.ReadFourByteInteger();
+                    }
+                    else if (propertyID == MqttPropertyID.AuthenticationMethod)
+                    {
+                        packet.Properties.AuthenticationMethod = body.ReadStringWithLengthPrefix();
+                    }
+                    else if (propertyID == MqttPropertyID.AuthenticationData)
+                    {
+                        packet.Properties.AuthenticationData = body.ReadWithLengthPrefix().ToArray();
+                    }
+                    else if (propertyID == MqttPropertyID.RetainAvailable)
+                    {
+                        packet.Properties.RetainAvailable = body.ReadBoolean();
+                    }
+                    else if (propertyID == MqttPropertyID.ReceiveMaximum)
+                    {
+                        packet.Properties.ReceiveMaximum = body.ReadTwoByteInteger();
+                    }
+                    else if (propertyID == MqttPropertyID.AssignedClientIdentifer)
+                    {
+                        packet.Properties.AssignedClientIdentifier = body.ReadStringWithLengthPrefix();
+                    }
+                    else if (propertyID == MqttPropertyID.TopicAliasMaximum)
+                    {
+                        packet.Properties.TopicAliasMaximum = body.ReadTwoByteInteger();
+                    }
+                    else if (propertyID == MqttPropertyID.ReasonString)
+                    {
+                        packet.Properties.ReasonString = body.ReadStringWithLengthPrefix();
+                    }
+                    else if (propertyID == MqttPropertyID.MaximumPacketSize)
+                    {
+                        packet.Properties.MaximumPacketSize = body.ReadFourByteInteger();
+                    }
+                    else if (propertyID == MqttPropertyID.WildcardSubscriptionAvailable)
+                    {
+                        packet.Properties.WildcardSubscriptionAvailable = body.ReadBoolean();
+                    }
+                    else if (propertyID == MqttPropertyID.SubscriptionIdentifierAvailable)
+                    {
+                        packet.Properties.SubscriptionIdentifiersAvailable = body.ReadBoolean();
+                    }
+                    else if (propertyID == MqttPropertyID.SharedSubscriptionAvailable)
+                    {
+                        packet.Properties.SharedSubscriptionAvailable = body.ReadBoolean();
+                    }
+                    else if (propertyID == MqttPropertyID.ServerKeepAlive)
+                    {
+                        packet.Properties.ServerKeepAlive = body.ReadTwoByteInteger();
+                    }
+                    else if (propertyID == MqttPropertyID.ResponseInformation)
+                    {
+                        packet.Properties.ResponseInformation = body.ReadStringWithLengthPrefix();
+                    }
+                    else if (propertyID == MqttPropertyID.ServerReference)
+                    {
+                        packet.Properties.ServerReference = body.ReadStringWithLengthPrefix();
+                    }
+                    else if (propertyID == MqttPropertyID.UserProperty)
+                    {
+                        packet.Properties.UserProperties = ReadUserProperties(body);
+                    }
+                }
+            }
+
+            return packet;
+        }
+
+        // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
+        private static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body)
+        {
+            if (body == null || body.Length == 0)
+            {
+                throw new MqttProtocolViolationException("Data from the body is required but not present.");
+            }
+        }
+
+        private static List<MqttUserProperty> ReadUserProperties(MqttPacketBodyReader reader)
+        {
+            var userPropertiesLength = reader.ReadVariableLengthInteger();
+            if (userPropertiesLength == 0)
+            {
+                return new List<MqttUserProperty>();
+            }
+
+            var userProperties = new List<MqttUserProperty>();
+            var targetPosition = reader.Offset + userPropertiesLength;
+            while (reader.Offset < targetPosition)
+            {
+                var name = reader.ReadStringWithLengthPrefix();
+                var value = reader.ReadStringWithLengthPrefix();
+
+                userProperties.Add(new MqttUserProperty(name, value));
+            }
+
+            return userProperties;
+        }
     }
 }

Source/MQTTnet/Formatter/V500/MqttV500PacketEncoder.cs

@@ -1,6 +1,223 @@
-namespace MQTTnet.Formatter.V500
+using System;
+using MQTTnet.Exceptions;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Formatter.V500
 {
     public class MqttV500PacketEncoder
     {
+        public byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
+        {
+            if (packet == null) throw new ArgumentNullException(nameof(packet));
+            if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
+
+            if (string.IsNullOrEmpty(packet.ClientId) && !packet.CleanSession)
+            {
+                throw new MqttProtocolViolationException("CleanSession must be set if ClientId is empty [MQTT-3.1.3-7].");
+            }
+
+            packetWriter.WriteWithLengthPrefix("MQTT");
+            packetWriter.Write(5); // [3.1.2.2 Protocol Version]
+
+            byte connectFlags = 0x0;
+            if (packet.CleanSession)
+            {
+                connectFlags |= 0x2;
+            }
+
+            if (packet.WillMessage != null)
+            {
+                connectFlags |= 0x4;
+                connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3);
+
+                if (packet.WillMessage.Retain)
+                {
+                    connectFlags |= 0x20;
+                }
+            }
+
+            if (packet.Password != null && packet.Username == null)
+            {
+                throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22].");
+            }
+
+            if (packet.Password != null)
+            {
+                connectFlags |= 0x40;
+            }
+
+            if (packet.Username != null)
+            {
+                connectFlags |= 0x80;
+            }
+
+            packetWriter.Write(connectFlags);
+            packetWriter.Write(packet.KeepAlivePeriod);
+
+            var propertiesWriter = new MqttV500PropertiesWriter();
+            if (packet.Properties != null)
+            {
+                propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.WillDelayInterval, packet.Properties.WillDelayInterval);
+                propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.SessionExpiryInterval, packet.Properties.SessionExpiryInterval);
+                propertiesWriter.Write(MqttPropertyID.AuthenticationMethod, packet.Properties.AuthenticationMethod);
+                propertiesWriter.Write(MqttPropertyID.AuthenticationData, packet.Properties.AuthenticationData);
+                propertiesWriter.Write(MqttPropertyID.RequestProblemInformation, packet.Properties.RequestProblemInformation);
+                propertiesWriter.Write(MqttPropertyID.RequestResponseInformation, packet.Properties.RequestResponseInformation);
+                propertiesWriter.Write(MqttPropertyID.ReceiveMaximum, packet.Properties.ReceiveMaximum);
+                propertiesWriter.Write(MqttPropertyID.TopicAlias, packet.Properties.TopicAliasMaximum);
+                propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.MaximumPacketSize, packet.Properties.MaximumPacketSize);
+                propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
+                propertiesWriter.WriteToPacket(packetWriter);
+            }
+
+            packetWriter.WriteWithLengthPrefix(packet.ClientId);
+
+            if (packet.WillMessage != null)
+            {
+                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic);
+                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload);
+            }
+
+            if (packet.Username != null)
+            {
+                packetWriter.WriteWithLengthPrefix(packet.Username);
+            }
+
+            if (packet.Password != null)
+            {
+                packetWriter.WriteWithLengthPrefix(packet.Password);
+            }
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
+        }
+
+        public byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
+        {
+            if (packet == null) throw new ArgumentNullException(nameof(packet));
+            if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
+
+            if (!packet.ConnectReasonCode.HasValue)
+            {
+                throw new MqttProtocolViolationException("The ConnectReasonCode must be set for MQTT version 5.");
+            }
+
+            byte connectAcknowledgeFlags = 0x0;
+            if (packet.IsSessionPresent)
+            {
+                connectAcknowledgeFlags |= 0x1;
+            }
+
+            packetWriter.Write(connectAcknowledgeFlags);
+            packetWriter.Write((byte)packet.ConnectReturnCode);
+            packetWriter.Write((byte)packet.ConnectReasonCode.Value);
+
+            var propertiesWriter = new MqttV500PropertiesWriter();
+            if (packet.Properties != null)
+            {
+                propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.SessionExpiryInterval, packet.Properties.SessionExpiryInterval);
+                propertiesWriter.Write(MqttPropertyID.AuthenticationMethod, packet.Properties.AuthenticationMethod);
+                propertiesWriter.Write(MqttPropertyID.AuthenticationData, packet.Properties.AuthenticationData);
+                propertiesWriter.Write(MqttPropertyID.RetainAvailable, packet.Properties.RetainAvailable);
+                propertiesWriter.Write(MqttPropertyID.ReceiveMaximum, packet.Properties.ReceiveMaximum);
+                propertiesWriter.Write(MqttPropertyID.AssignedClientIdentifer, packet.Properties.AssignedClientIdentifier);
+                propertiesWriter.Write(MqttPropertyID.TopicAliasMaximum, packet.Properties.TopicAliasMaximum);
+                propertiesWriter.Write(MqttPropertyID.ReasonString, packet.Properties.ReasonString);
+                propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.MaximumPacketSize, packet.Properties.MaximumPacketSize);
+                propertiesWriter.Write(MqttPropertyID.WildcardSubscriptionAvailable, packet.Properties.WildcardSubscriptionAvailable);
+                propertiesWriter.Write(MqttPropertyID.SubscriptionIdentifierAvailable, packet.Properties.SubscriptionIdentifiersAvailable);
+                propertiesWriter.Write(MqttPropertyID.SharedSubscriptionAvailable, packet.Properties.SharedSubscriptionAvailable);
+                propertiesWriter.Write(MqttPropertyID.ServerKeepAlive, packet.Properties.ServerKeepAlive);
+                propertiesWriter.Write(MqttPropertyID.ResponseInformation, packet.Properties.ResponseInformation);
+                propertiesWriter.Write(MqttPropertyID.ServerReference, packet.Properties.ServerReference);
+                propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
+                propertiesWriter.WriteToPacket(packetWriter);
+            }
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
+        }
+
+        public byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
+        {
+            if (packet == null) throw new ArgumentNullException(nameof(packet));
+            if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
+
+            if (packet.QualityOfServiceLevel == 0 && packet.Dup)
+            {
+                throw new MqttProtocolViolationException("Dup flag must be false for QoS 0 packets [MQTT-3.3.1-2].");
+            }
+
+            packetWriter.WriteWithLengthPrefix(packet.Topic);
+
+            if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
+            {
+                if (!packet.PacketIdentifier.HasValue)
+                {
+                    throw new MqttProtocolViolationException("Publish packet has no packet identifier.");
+                }
+
+                packetWriter.Write(packet.PacketIdentifier.Value);
+            }
+            else
+            {
+                if (packet.PacketIdentifier > 0)
+                {
+                    throw new MqttProtocolViolationException("Packet identifier must be empty if QoS == 0 [MQTT-2.3.1-5].");
+                }
+            }
+
+            var propertiesWriter = new MqttV500PropertiesWriter();
+            propertiesWriter.Write(MqttPropertyID.PayloadFormatIndicator, packet.Properties.PayloadFormatIndicator);
+            propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.MessageExpiryInterval, packet.Properties.MessageExpiryInterval);
+            propertiesWriter.Write(MqttPropertyID.TopicAlias, packet.Properties.TopicAlias);
+            propertiesWriter.Write(MqttPropertyID.ResponseTopic, packet.Properties.ResponseTopic);
+            propertiesWriter.Write(MqttPropertyID.CorrelationData, packet.Properties.CorrelationData);
+            propertiesWriter.WriteAsVariableLengthInteger(MqttPropertyID.SubscriptionIdentifier, packet.Properties.SubscriptionIdentifier);
+            propertiesWriter.Write(MqttPropertyID.ContentType, packet.Properties.ContentType);
+            propertiesWriter.WriteUserProperties(packet.Properties.UserProperties);
+            propertiesWriter.WriteToPacket(packetWriter);
+
+            if (packet.Payload?.Length > 0)
+            {
+                packetWriter.Write(packet.Payload, 0, packet.Payload.Length);
+            }
+
+            byte fixedHeader = 0;
+
+            if (packet.Retain)
+            {
+                fixedHeader |= 0x01;
+            }
+
+            fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1);
+
+            if (packet.Dup)
+            {
+                fixedHeader |= 0x08;
+            }
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
+        }
+
+        public byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
+        {
+            if (packet == null) throw new ArgumentNullException(nameof(packet));
+            if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
+
+            if (!packet.PacketIdentifier.HasValue)
+            {
+                throw new MqttProtocolViolationException("PubAck packet has no packet identifier.");
+            }
+
+            if (!packet.ConnectReasonCode.HasValue)
+            {
+                throw new MqttProtocolViolationException("PubAck packet must contain a connect reason.");
+            }
+
+            packetWriter.Write(packet.PacketIdentifier.Value);
+            packetWriter.Write((byte)packet.ConnectReasonCode.Value);
+
+            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
+        }
     }
 }

Source/MQTTnet/Formatter/V500/MqttV500PacketFormatter.cs

@@ -1,10 +1,6 @@
 using System;
-using System.Collections.Generic;
-using MQTTnet.Exceptions;
 using MQTTnet.Formatter.V311;
 using MQTTnet.Packets;
-using MQTTnet.Packets.Properties;
-using MQTTnet.Protocol;
 
 namespace MQTTnet.Formatter.V500
 {
@@ -13,184 +9,47 @@ namespace MQTTnet.Formatter.V500
         private readonly MqttV500PacketEncoder _encoder = new MqttV500PacketEncoder();
         private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder();
 
+        public override MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+
+            return new MqttPublishPacket
+            {
+                Topic = applicationMessage.Topic,
+                Payload = applicationMessage.Payload,
+                QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel,
+                Retain = applicationMessage.Retain,
+                Dup = false,
+                Properties = new MqttPublishPacketProperties
+                {
+                    UserProperties = applicationMessage.UserProperties
+                }
+            };
+        }
+
         protected override byte SerializeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
         {
-            ValidateConnectPacket(packet);
-
-            packetWriter.WriteWithLengthPrefix("MQTT");
-            packetWriter.Write(5); // [3.1.2.2 Protocol Version]
-
-            byte connectFlags = 0x0;
-            if (packet.CleanSession)
-            {
-                connectFlags |= 0x2;
-            }
-
-            if (packet.WillMessage != null)
-            {
-                connectFlags |= 0x4;
-                connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3);
-
-                if (packet.WillMessage.Retain)
-                {
-                    connectFlags |= 0x20;
-                }
-            }
-
-            if (packet.Password != null && packet.Username == null)
-            {
-                throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22].");
-            }
-
-            if (packet.Password != null)
-            {
-                connectFlags |= 0x40;
-            }
-
-            if (packet.Username != null)
-            {
-                connectFlags |= 0x80;
-            }
-
-            packetWriter.Write(connectFlags);
-            packetWriter.Write(packet.KeepAlivePeriod);
-            packetWriter.WriteWithLengthPrefix(packet.ClientId);
-
-            if (packet.WillMessage != null)
-            {
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic);
-                packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload);
-            }
-
-            if (packet.Username != null)
-            {
-                packetWriter.WriteWithLengthPrefix(packet.Username);
-            }
-
-            if (packet.Password != null)
-            {
-                packetWriter.WriteWithLengthPrefix(packet.Password);
-            }
-
-            WriteProperty(MqttMessagePropertyID.WillDelayInterval, packet.WillDelayIntervalProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.SessionExpiryInterval, packet.SessionExpiryIntervalProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.AuthenticationMethod, packet.AuthenticationMethodProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.AuthenticationData, packet.AuthenticationDataProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.RequestProblemInformation, packet.RequestProblemInformationProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.RequestResponseInformation, packet.RequestResponseInformationProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.ReceiveMaximum, packet.ReceiveMaximumProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.TopicAlias, packet.TopicAliasMaximumProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.MaximumPacketSize, packet.MaximumPacketSizeProperty, packetWriter);
-            WriteUserProperties(packet.UserPropertiesProperty, packetWriter);
-
-            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
+            return _encoder.EncodeConnectPacket(packet, packetWriter);
         }
 
         protected override byte SerializeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
         {
-            byte connectAcknowledgeFlags = 0x0;
-            if (packet.IsSessionPresent)
-            {
-                connectAcknowledgeFlags |= 0x1;
-            }
-
-            packetWriter.Write(connectAcknowledgeFlags);
-            packetWriter.Write((byte)packet.ConnectReturnCode);
-
-            WriteProperty(MqttMessagePropertyID.SessionExpiryInterval, packet.SessionExpiryIntervalProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.AuthenticationMethod, packet.AuthenticationMethodProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.AuthenticationData, packet.AuthenticationDataProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.RetainAvailable, packet.RetainAvailableProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.ReceiveMaximum, packet.ReceiveMaximumProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.AssignedClientIdentifer, packet.AssignedClientIdentifierProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.TopicAlias, packet.TopicAliasMaximumProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.ReasonString, packet.ReasonStringProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.MaximumPacketSize, packet.MaximumPacketSizeProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.WildcardSubscriptionAvailable, packet.WildcardSubscriptionAvailableProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.SubscriptionIdentifierAvailable, packet.SubscriptionIdentifiersAvailableProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.SharedSubscriptionAvailable, packet.SharedSubscriptionAvailableProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.ServerKeepAlive, packet.ServerKeepAliveProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.ResponseInformation, packet.ResponseInformationProperty, packetWriter);
-            WriteProperty(MqttMessagePropertyID.ServerReference, packet.ServerReferenceProperty, packetWriter);
-            WriteUserProperties(packet.UserPropertiesProperty, packetWriter);
-
-            return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
+            return _encoder.EncodeConnAckPacket(packet, packetWriter);
         }
 
-        private static void WriteProperty(MqttMessagePropertyID id, bool? value, MqttPacketWriter target)
+        protected override MqttBasePacket DeserializeConnAckPacket(MqttPacketBodyReader body)
         {
-            if (!value.HasValue)
-            {
-                return;
-            }
-
-            target.Write((byte)id);
-            target.Write(value.Value ? (byte)0x1 : (byte)0x0);
+            return _decoder.DecodeConnAckPacket(body);
         }
 
-        private static void WriteProperty(MqttMessagePropertyID id, ushort? value, MqttPacketWriter target)
+        protected override byte SerializePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
         {
-            if (!value.HasValue)
-            {
-                return;
-            }
-
-            target.Write((byte)id);
-            target.Write(value.Value);
+            return _encoder.EncodePublishPacket(packet, packetWriter);
         }
 
-        private static void WriteProperty(MqttMessagePropertyID id, uint? value, MqttPacketWriter target)
+        protected override byte SerializePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
         {
-            if (!value.HasValue)
-            {
-                return;
-            }
-
-            // TODO: Check if order must be reversed like for ushort.
-            target.Write((byte)id);
-            var bytes = BitConverter.GetBytes(value.Value);
-            target.Write(bytes, 0, bytes.Length);
-        }
-
-        private static void WriteProperty(MqttMessagePropertyID id, string value, MqttPacketWriter target)
-        {
-            if (value == null)
-            {
-                return;
-            }
-            
-            target.Write((byte)id);
-            target.WriteWithLengthPrefix(value);
-        }
-
-        private static void WriteProperty(MqttMessagePropertyID id, byte[] value, MqttPacketWriter target)
-        {
-            if (value == null)
-            {
-                return;
-            }
-
-            target.Write((byte)id);
-            target.WriteWithLengthPrefix(value);
-        }
-
-        private static void WriteUserProperties(List<MqttUserProperty> userProperties, MqttPacketWriter target)
-        {
-            if (userProperties == null || userProperties.Count == 0)
-            {
-                return;
-            }
-
-            var propertyWriter = new MqttPacketWriter();
-            foreach (var property in userProperties)
-            {
-                propertyWriter.WriteWithLengthPrefix(property.Name);
-                propertyWriter.WriteWithLengthPrefix(property.Value);
-            }
-
-            target.Write((byte)MqttMessagePropertyID.UserProperty);
-            target.WriteVariableLengthInteger((uint)propertyWriter.Length);
-            target.Write(propertyWriter);
+            return _encoder.EncodePubAckPacket(packet, packetWriter);
         }
     }
 }

Source/MQTTnet/Formatter/V500/MqttV500PropertiesWriter.cs

@@ -0,0 +1,108 @@
+using System;
+using System.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Formatter.V500
+{
+    public class MqttV500PropertiesWriter
+    {
+        private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
+
+        public void Write(MqttPropertyID id, bool? value)
+        {
+            if (!value.HasValue)
+            {
+                return;
+            }
+
+            _packetWriter.Write((byte)id);
+            _packetWriter.Write(value.Value ? (byte)0x1 : (byte)0x0);
+        }
+
+        public void Write(MqttPropertyID id, ushort? value)
+        {
+            if (!value.HasValue)
+            {
+                return;
+            }
+
+            _packetWriter.Write((byte)id);
+            _packetWriter.Write(value.Value);
+        }
+
+        public void WriteAsVariableLengthInteger(MqttPropertyID id, uint? value)
+        {
+            if (!value.HasValue)
+            {
+                return;
+            }
+
+            _packetWriter.Write((byte)id);
+            _packetWriter.WriteVariableLengthInteger(value.Value);
+        }
+
+        public void WriteAsFourByteInteger(MqttPropertyID id, uint? value)
+        {
+            if (!value.HasValue)
+            {
+                return;
+            }
+
+            _packetWriter.Write((byte)id);
+            _packetWriter.Write((byte)(value.Value >> 24));
+            _packetWriter.Write((byte)(value.Value >> 16));
+            _packetWriter.Write((byte)(value.Value >> 8));
+            _packetWriter.Write((byte)value.Value);
+        }
+
+        public void Write(MqttPropertyID id, string value)
+        {
+            if (value == null)
+            {
+                return;
+            }
+
+            _packetWriter.Write((byte)id);
+            _packetWriter.WriteWithLengthPrefix(value);
+        }
+
+        public void Write(MqttPropertyID id, byte[] value)
+        {
+            if (value == null)
+            {
+                return;
+            }
+
+            _packetWriter.Write((byte)id);
+            _packetWriter.WriteWithLengthPrefix(value);
+        }
+
+        public void WriteUserProperties(List<MqttUserProperty> userProperties)
+        {
+            if (userProperties == null || userProperties.Count == 0)
+            {
+                return;
+            }
+
+            var propertyWriter = new MqttPacketWriter();
+            foreach (var property in userProperties)
+            {
+                propertyWriter.WriteWithLengthPrefix(property.Name);
+                propertyWriter.WriteWithLengthPrefix(property.Value);
+            }
+
+            _packetWriter.Write((byte)MqttPropertyID.UserProperty);
+            _packetWriter.WriteVariableLengthInteger((uint)propertyWriter.Length);
+            _packetWriter.Write(propertyWriter);
+        }
+
+        public void WriteToPacket(MqttPacketWriter packetWriter)
+        {
+            if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
+
+            packetWriter.WriteVariableLengthInteger((uint)_packetWriter.Length);
+            packetWriter.Write(packetWriter);
+        }
+    }
+}