remove async and framing from serializer

MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs

@@ -1,4 +1,5 @@
 using System;
+using System.IO;
 using System.Threading.Tasks;
 using MQTTnet.Core.Channel;
 using MQTTnet.Core.Client;
@@ -35,21 +36,33 @@ namespace MQTTnet.Core.Adapter
         {
             MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), "TX >>> {0} [Timeout={1}]", packet, timeout);
 
-            return ExecuteWithTimeoutAsync(PacketSerializer.SerializeAsync(packet, _channel), timeout);
+            var writeBuffer = PacketSerializer.Serialize(packet);
+            _sendTask = SendAsync( writeBuffer );
+            return ExecuteWithTimeoutAsync(_sendTask, timeout);
+        }
+
+        private Task _sendTask = Task.FromResult(0); // this task is used to prevent overlapping write
+
+        private async Task SendAsync(byte[] buffer)
+        {
+            await _sendTask.ConfigureAwait(false);
+            await _channel.Stream.WriteAsync(buffer, 0, buffer.Length).ConfigureAwait( false );
         }
 
         public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout)
         {
-            MqttBasePacket packet;
+            Tuple<MqttPacketHeader, MemoryStream> tuple;
             if (timeout > TimeSpan.Zero)
             {
-                packet = await ExecuteWithTimeoutAsync(PacketSerializer.DeserializeAsync(_channel), timeout).ConfigureAwait(false);
+                tuple = await ExecuteWithTimeoutAsync(ReceiveAsync(), timeout).ConfigureAwait(false);
             }
             else
             {
-                packet = await PacketSerializer.DeserializeAsync(_channel).ConfigureAwait(false);
+                tuple = await ReceiveAsync().ConfigureAwait(false);
             }
 
+            var packet = PacketSerializer.Deserialize(tuple.Item1, tuple.Item2);
+
             if (packet == null)
             {
                 throw new MqttProtocolViolationException("Received malformed packet.");
@@ -59,6 +72,31 @@ namespace MQTTnet.Core.Adapter
             return packet;
         }
 
+        private async Task<Tuple<MqttPacketHeader, MemoryStream>> ReceiveAsync()
+        {
+            var header = MqttPacketReader.ReadHeaderFromSource(_channel.Stream);
+
+            MemoryStream body = null;
+            if (header.BodyLength > 0)
+            {
+                var totalRead = 0;
+                var readBuffer = new byte[header.BodyLength];
+                do
+                {
+                    var read = await _channel.Stream.ReadAsync(readBuffer, totalRead, header.BodyLength - totalRead)
+                        .ConfigureAwait( false );
+                    totalRead += read;
+                } while (totalRead < header.BodyLength);
+                body = new MemoryStream(readBuffer, 0, header.BodyLength);
+            }
+            else
+            {
+                body = new MemoryStream();
+            }
+
+            return Tuple.Create(header, body);
+        }
+
         private static async Task<TResult> ExecuteWithTimeoutAsync<TResult>(Task<TResult> task, TimeSpan timeout)
         {
             var timeoutTask = Task.Delay(timeout);

MQTTnet.Core/Serializer/IMqttPacketSerializer.cs

@@ -1,5 +1,4 @@
-using System.Threading.Tasks;
+using System.IO;
-using MQTTnet.Core.Channel;
 using MQTTnet.Core.Packets;
 
 namespace MQTTnet.Core.Serializer
@@ -8,8 +7,8 @@ namespace MQTTnet.Core.Serializer
     {
         MqttProtocolVersion ProtocolVersion { get; set; }
 
-        Task SerializeAsync(MqttBasePacket mqttPacket, IMqttCommunicationChannel destination);
+        byte[] Serialize(MqttBasePacket mqttPacket);
 
-        Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source);
+        MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream stream);
     }
 }

MQTTnet.Core/Serializer/MqttPacketReader.cs

@@ -14,7 +14,7 @@ namespace MQTTnet.Core.Serializer
         private readonly MqttPacketHeader _header;
 
         public MqttPacketReader(Stream stream, MqttPacketHeader header)
-            : base(stream)
+            : base(stream, Encoding.UTF8, true)
         {
             _header = header;
         }
@@ -49,13 +49,13 @@ namespace MQTTnet.Core.Serializer
             return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
         }
 
-        public static MqttPacketHeader ReadHeaderFromSource(IMqttCommunicationChannel source)
+        public static MqttPacketHeader ReadHeaderFromSource(Stream stream)
         {
-            var fixedHeader = (byte)source.Stream.ReadByte();
+            var fixedHeader = (byte)stream.ReadByte();
             var byteReader = new ByteReader(fixedHeader);
             byteReader.Read(4);
             var controlPacketType = (MqttControlPacketType)byteReader.Read(4);
-            var bodyLength = ReadBodyLengthFromSource(source);
+            var bodyLength = ReadBodyLengthFromSource(stream);
 
             return new MqttPacketHeader()
             {
@@ -65,7 +65,7 @@ namespace MQTTnet.Core.Serializer
             };
         }
         
-        private static int ReadBodyLengthFromSource(IMqttCommunicationChannel source)
+        private static int ReadBodyLengthFromSource(Stream stream)
         {
             // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
             var multiplier = 1;
@@ -73,7 +73,7 @@ namespace MQTTnet.Core.Serializer
             byte encodedByte;
             do
             {
-                encodedByte = (byte)source.Stream.ReadByte();
+                encodedByte = (byte)stream.ReadByte();
                 value += (encodedByte & 127) * multiplier;
                 multiplier *= 128;
                 if (multiplier > 128 * 128 * 128)

MQTTnet.Core/Serializer/MqttPacketSerializer.cs

@@ -17,12 +17,10 @@ namespace MQTTnet.Core.Serializer
         private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs");
 
         public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
-        private Task _sendTask = Task.FromResult( 0 );
 
-        public async Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
+        public byte[] Serialize(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
-            if (destination == null) throw new ArgumentNullException(nameof(destination));
 
             using (var stream = new MemoryStream())
             using (var writer = new MqttPacketWriter(stream))
@@ -37,17 +35,10 @@ namespace MQTTnet.Core.Serializer
                 Buffer.BlockCopy( headerArray, 0, writeBuffer, 0, headerArray.Length );
                 Buffer.BlockCopy( body, 0, writeBuffer, headerArray.Length, body.Length );
 
-                _sendTask = Send( writeBuffer, destination );
+                return writeBuffer;
-                await _sendTask.ConfigureAwait( false );
             }
         }
 
-        private async Task Send(byte[] buffer, IMqttCommunicationChannel destination )
-        {
-            await _sendTask.ConfigureAwait( false );
-            await destination.Stream.WriteAsync( buffer, 0, buffer.Length ).ConfigureAwait( false );
-        }
-
         private byte SerializePacket(MqttBasePacket packet, MqttPacketWriter writer)
         {
             if (packet is MqttConnectPacket connectPacket)
@@ -123,29 +114,11 @@ namespace MQTTnet.Core.Serializer
             throw new MqttProtocolViolationException("Packet type invalid.");
         }
 
-        public async Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source)
+        public MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body)
         {
-            if (source == null) throw new ArgumentNullException(nameof(source));
+            if (header == null) throw new ArgumentNullException(nameof(header));
+            if (body == null) throw new ArgumentNullException(nameof(body));
             
-            var header = MqttPacketReader.ReadHeaderFromSource(source);
-            
-            MemoryStream body = null;
-            if ( header.BodyLength > 0 )
-            {
-                var totalRead = 0;
-                var readBuffer = new byte[header.BodyLength];
-                do
-                {
-                    var read = await source.Stream.ReadAsync( readBuffer, totalRead, header.BodyLength - totalRead ).ConfigureAwait(false);
-                    totalRead += read;
-                } while ( totalRead < header.BodyLength );
-                body = new MemoryStream( readBuffer, 0, header.BodyLength );
-            }
-            else
-            {
-                body = new MemoryStream();
-            }
-
             using (var reader = new MqttPacketReader(body, header))
             {
                 return Deserialize( header, reader );

Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs

@@ -424,9 +424,7 @@ namespace MQTTnet.Core.Tests
         private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
         {
             var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion };
-            var channel = new TestChannel();
+            var buffer = serializer.Serialize(packet);
-            serializer.SerializeAsync(packet, channel).Wait();
-            var buffer = channel.ToArray();
 
             Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer));
         }
@@ -434,19 +432,21 @@ namespace MQTTnet.Core.Tests
         private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
         {
             var serializer = new MqttPacketSerializer();
+            
+            var buffer1 = serializer.Serialize(packet);
 
-            var channel1 = new TestChannel();
+            using (var headerStream = new MemoryStream( buffer1 ))
-            serializer.SerializeAsync(packet, channel1).Wait();
+            {
-            var buffer1 = channel1.ToArray();
+                var header = MqttPacketReader.ReadHeaderFromSource( headerStream );
 
-            var channel2 = new TestChannel(buffer1);
+                using (var bodyStream = new MemoryStream( buffer1, (int)headerStream.Position, header.BodyLength ))
-            var deserializedPacket = serializer.DeserializeAsync(channel2).Result;
+                {
-            var buffer2 = channel2.ToArray();
+                    var deserializedPacket = serializer.Deserialize(header, bodyStream);
+                    var buffer2 = serializer.Serialize( deserializedPacket );
 
-            var channel3 = new TestChannel(buffer2);
+                    Assert.AreEqual( expectedBase64Value, Convert.ToBase64String( buffer2 ) );
-            serializer.SerializeAsync(deserializedPacket, channel3).Wait();
+                } 
-
+            }
-            Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(channel3.ToArray()));
         }
     }
 }