From c001897590e8a6acec52c9cbb74ef8749d3d5311 Mon Sep 17 00:00:00 2001 From: Eggers Jan Date: Mon, 11 Sep 2017 13:34:06 +0200 Subject: [PATCH] remove async and framing from serializer --- .../MqttChannelCommunicationAdapter.cs | 46 +++++++++++++++++-- .../Serializer/IMqttPacketSerializer.cs | 7 ++- MQTTnet.Core/Serializer/MqttPacketReader.cs | 12 ++--- .../Serializer/MqttPacketSerializer.cs | 37 ++------------- .../MqttPacketSerializerTests.cs | 26 +++++------ 5 files changed, 69 insertions(+), 59 deletions(-) diff --git a/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs b/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs index d0f1c6e0..e42b5e52 100644 --- a/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs +++ b/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 ReceivePacketAsync(TimeSpan timeout) { - MqttBasePacket packet; + Tuple 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> 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 ExecuteWithTimeoutAsync(Task task, TimeSpan timeout) { var timeoutTask = Task.Delay(timeout); diff --git a/MQTTnet.Core/Serializer/IMqttPacketSerializer.cs b/MQTTnet.Core/Serializer/IMqttPacketSerializer.cs index 801b7ea6..df5b045b 100644 --- a/MQTTnet.Core/Serializer/IMqttPacketSerializer.cs +++ b/MQTTnet.Core/Serializer/IMqttPacketSerializer.cs @@ -1,5 +1,4 @@ -using System.Threading.Tasks; -using MQTTnet.Core.Channel; +using System.IO; 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 DeserializeAsync(IMqttCommunicationChannel source); + MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream stream); } } \ No newline at end of file diff --git a/MQTTnet.Core/Serializer/MqttPacketReader.cs b/MQTTnet.Core/Serializer/MqttPacketReader.cs index ea479c28..10650781 100644 --- a/MQTTnet.Core/Serializer/MqttPacketReader.cs +++ b/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) diff --git a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs b/MQTTnet.Core/Serializer/MqttPacketSerializer.cs index 78c3cc0a..fa373f06 100644 --- a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs +++ b/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 ); - await _sendTask.ConfigureAwait( false ); + return writeBuffer; } } - 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 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 ); diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index 4f33be33..bf0d6f5d 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/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(); - serializer.SerializeAsync(packet, channel).Wait(); - var buffer = channel.ToArray(); + var buffer = serializer.Serialize(packet); 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(); - serializer.SerializeAsync(packet, channel1).Wait(); - var buffer1 = channel1.ToArray(); + using (var headerStream = new MemoryStream( buffer1 )) + { + var header = MqttPacketReader.ReadHeaderFromSource( headerStream ); - var channel2 = new TestChannel(buffer1); - var deserializedPacket = serializer.DeserializeAsync(channel2).Result; - var buffer2 = channel2.ToArray(); + using (var bodyStream = new MemoryStream( buffer1, (int)headerStream.Position, header.BodyLength )) + { + var deserializedPacket = serializer.Deserialize(header, bodyStream); + var buffer2 = serializer.Serialize( deserializedPacket ); - var channel3 = new TestChannel(buffer2); - serializer.SerializeAsync(deserializedPacket, channel3).Wait(); - - Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(channel3.ToArray())); + Assert.AreEqual( expectedBase64Value, Convert.ToBase64String( buffer2 ) ); + } + } } } }