Improve payload handling and allocation.

Source/MQTTnet/Formatter/MqttPacketWriter.cs

@@ -79,24 +79,21 @@ namespace MQTTnet.Formatter
             else
             {
                 WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value));
-            }         
+            }
         }
 
         public void WriteWithLengthPrefix(byte[] value)
         {
-            if (value == null) throw new ArgumentNullException(nameof(value));
-
-            EnsureAdditionalCapacity(value.Length + 2);
-
-            if (value.Length == 0)
+            if (value == null || value.Length == 0)
             {
-                Write(ZeroTwoByteIntegerArray);   
+                Write(ZeroTwoByteIntegerArray);
             }
             else
             {
+                EnsureAdditionalCapacity(value.Length + 2);
                 Write((ushort)value.Length);
                 Write(value, 0, value.Length);
-            }           
+            }
         }
 
         public void Write(byte @byte)
@@ -131,7 +128,7 @@ namespace MQTTnet.Formatter
             Array.Copy(buffer, offset, _buffer, _offset, count);
             IncreasePosition(count);
         }
-        
+
         public void Write(MqttPacketWriter propertyWriter)
         {
             if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));

Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs

@@ -56,17 +56,17 @@ namespace MQTTnet.Formatter.V3
                 case MqttControlPacketType.Connect: return DecodeConnectPacket(receivedMqttPacket.Body);
                 case MqttControlPacketType.ConnAck: return DecodeConnAckPacket(receivedMqttPacket.Body);
                 case MqttControlPacketType.Disconnect: return new MqttDisconnectPacket();
-                case MqttControlPacketType.Publish: return DecodePublish(receivedMqttPacket);
-                case MqttControlPacketType.PubAck: return DecodePubAck(receivedMqttPacket.Body);
-                case MqttControlPacketType.PubRec: return DecodePubRec(receivedMqttPacket.Body);
-                case MqttControlPacketType.PubRel: return DecodePubRel(receivedMqttPacket.Body);
-                case MqttControlPacketType.PubComp: return DecodePubComp(receivedMqttPacket.Body);
+                case MqttControlPacketType.Publish: return DecodePublishPacket(receivedMqttPacket);
+                case MqttControlPacketType.PubAck: return DecodePubAckPacket(receivedMqttPacket.Body);
+                case MqttControlPacketType.PubRec: return DecodePubRecPacket(receivedMqttPacket.Body);
+                case MqttControlPacketType.PubRel: return DecodePubRelPacket(receivedMqttPacket.Body);
+                case MqttControlPacketType.PubComp: return DecodePubCompPacket(receivedMqttPacket.Body);
                 case MqttControlPacketType.PingReq: return new MqttPingReqPacket();
                 case MqttControlPacketType.PingResp: return new MqttPingRespPacket();
-                case MqttControlPacketType.Subscribe: return DecodeSubscribe(receivedMqttPacket.Body);
-                case MqttControlPacketType.SubAck: return DecodeSubAck(receivedMqttPacket.Body);
-                case MqttControlPacketType.Unsubscibe: return DecodeUnsubscribe(receivedMqttPacket.Body);
-                case MqttControlPacketType.UnsubAck: return DecodeUnsubAck(receivedMqttPacket.Body);
+                case MqttControlPacketType.Subscribe: return DecodeSubscribePacket(receivedMqttPacket.Body);
+                case MqttControlPacketType.SubAck: return DecodeSubAckPacket(receivedMqttPacket.Body);
+                case MqttControlPacketType.Unsubscibe: return DecodeUnsubscribePacket(receivedMqttPacket.Body);
+                case MqttControlPacketType.UnsubAck: return DecodeUnsubAckPacket(receivedMqttPacket.Body);
 
                 default: throw new MqttProtocolViolationException($"Packet type ({controlPacketType}) not supported.");
             }
@@ -100,7 +100,7 @@ namespace MQTTnet.Formatter.V3
             }
         }
 
-        private static MqttBasePacket DecodeUnsubAck(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeUnsubAckPacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -110,7 +110,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubComp(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubCompPacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -120,7 +120,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubRel(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubRelPacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -130,7 +130,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubRec(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubRecPacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -140,7 +140,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodePubAck(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodePubAckPacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -150,7 +150,7 @@ namespace MQTTnet.Formatter.V3
             };
         }
 
-        private static MqttBasePacket DecodeUnsubscribe(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeUnsubscribePacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -167,7 +167,7 @@ namespace MQTTnet.Formatter.V3
             return packet;
         }
 
-        private static MqttBasePacket DecodeSubscribe(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeSubscribePacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 
@@ -190,7 +190,7 @@ namespace MQTTnet.Formatter.V3
             return packet;
         }
 
-        private static MqttBasePacket DecodePublish(ReceivedMqttPacket receivedMqttPacket)
+        private static MqttBasePacket DecodePublishPacket(ReceivedMqttPacket receivedMqttPacket)
         {
             ThrowIfBodyIsEmpty(receivedMqttPacket.Body);
 
@@ -211,11 +211,15 @@ namespace MQTTnet.Formatter.V3
                 PacketIdentifier = packetIdentifier,
                 Retain = retain,
                 Topic = topic,
-                Payload = receivedMqttPacket.Body.ReadRemainingData().ToArray(),
                 QualityOfServiceLevel = qualityOfServiceLevel,
                 Dup = dup
             };
 
+            if (!receivedMqttPacket.Body.EndOfStream)
+            {
+                packet.Payload = receivedMqttPacket.Body.ReadRemainingData().ToArray();
+            }
+
             return packet;
         }
 
@@ -271,7 +275,7 @@ namespace MQTTnet.Formatter.V3
             return packet;
         }
 
-        private static MqttBasePacket DecodeSubAck(MqttPacketBodyReader body)
+        private static MqttBasePacket DecodeSubAckPacket(MqttPacketBodyReader body)
         {
             ThrowIfBodyIsEmpty(body);
 

Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs

@@ -503,7 +503,10 @@ namespace MQTTnet.Formatter.V5
                 }
             }
 
-            packet.Payload = body.ReadRemainingData().ToArray();
+            if (!body.EndOfStream)
+            {
+                packet.Payload = body.ReadRemainingData().ToArray();
+            }
 
             return packet;
         }

Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs

@@ -380,7 +380,7 @@ namespace MQTTnet.Formatter.V5
 
         private static byte EncodeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter)
         {
-            if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
+            if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
 
             ThrowIfPacketIdentifierIsInvalid(packet);
 
@@ -427,12 +427,9 @@ namespace MQTTnet.Formatter.V5
 
         private static byte EncodeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter)
         {
-            ThrowIfPacketIdentifierIsInvalid(packet);
+            if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
 
-            if (packet.ReasonCodes == null || !packet.ReasonCodes.Any())
-            {
-                throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
-            }
+            ThrowIfPacketIdentifierIsInvalid(packet);
 
             packetWriter.Write(packet.PacketIdentifier.Value);
 
@@ -455,7 +452,7 @@ namespace MQTTnet.Formatter.V5
 
         private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter)
         {
-            if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
+            if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
 
             ThrowIfPacketIdentifierIsInvalid(packet);
 
@@ -479,13 +476,10 @@ namespace MQTTnet.Formatter.V5
 
         private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter)
         {
+            if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
+
             ThrowIfPacketIdentifierIsInvalid(packet);
-
-            if (packet.ReasonCodes == null || !packet.ReasonCodes.Any())
-            {
-                throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
-            }
-
+            
             packetWriter.Write(packet.PacketIdentifier.Value);
 
             var propertiesWriter = new MqttV500PropertiesWriter();