diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs index a64b4451..47dfb453 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs @@ -10,6 +10,7 @@ using MQTTnet.Core.Adapter; using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Serializer; using MQTTnet.Core.Server; +using MQTTnet.Core.Channel; namespace MQTTnet.Implementations { @@ -86,7 +87,7 @@ namespace MQTTnet.Implementations try { var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false); - var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer()); + var clientAdapter = new MqttChannelCommunicationAdapter(new BufferedCommunicationChannel(new MqttTcpChannel(clientSocket, null)), new MqttPacketSerializer()); ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter)); } catch (Exception exception) when (!(exception is ObjectDisposedException)) diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs index 5626256c..66fc2938 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs @@ -93,17 +93,15 @@ namespace MQTTnet.Implementations } } - public async Task ReadAsync(byte[] buffer) + public async Task> ReadAsync(int length, byte[] buffer) { - if (buffer == null) throw new ArgumentNullException(nameof(buffer)); - try { var totalBytes = 0; do { - var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes).ConfigureAwait(false); + var read = await _dataStream.ReadAsync(buffer, totalBytes, length - totalBytes).ConfigureAwait(false); if (read == 0) { throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting)); @@ -111,7 +109,8 @@ namespace MQTTnet.Implementations totalBytes += read; } - while (totalBytes < buffer.Length); + while (totalBytes < length); + return new ArraySegment(buffer, 0, length); } catch (SocketException exception) { @@ -143,5 +142,10 @@ namespace MQTTnet.Implementations return certificates; } + + public int Peek() + { + return _socket.Available; + } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs index b90b02a8..94f1330f 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs @@ -11,9 +11,6 @@ namespace MQTTnet.Implementations public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable { private ClientWebSocket _webSocket = new ClientWebSocket(); - private const int BufferSize = 4096; - private const int BufferAmplifier = 20; - private readonly byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier]; private int WebSocketBufferSize; private int WebSocketBufferOffset; @@ -42,51 +39,40 @@ namespace MQTTnet.Implementations _webSocket?.Dispose(); } - public Task ReadAsync(byte[] buffer) + public async Task> ReadAsync(int length, byte[] buffer) { - return Task.WhenAll(ReadToBufferAsync(buffer)); + await ReadToBufferAsync(length, buffer).ConfigureAwait(false); + + var result = new ArraySegment(buffer, WebSocketBufferOffset, length); + WebSocketBufferSize -= length; + WebSocketBufferOffset += length; + + return result; } - private async Task ReadToBufferAsync(byte[] buffer) + private async Task ReadToBufferAsync(int length, byte[] buffer) { - var temporaryBuffer = new byte[BufferSize]; - var offset = 0; - - while (_webSocket.State == WebSocketState.Open) + if (WebSocketBufferSize > 0) { - if (WebSocketBufferSize == 0) - { - WebSocketBufferOffset = 0; - - WebSocketReceiveResult response; - do - { - response = await _webSocket.ReceiveAsync(new ArraySegment(temporaryBuffer), CancellationToken.None).ConfigureAwait(false); - - temporaryBuffer.CopyTo(WebSocketBuffer, offset); - offset += response.Count; - temporaryBuffer = new byte[BufferSize]; - } while (!response.EndOfMessage); - - WebSocketBufferSize = response.Count; - if (response.MessageType == WebSocketMessageType.Close) - { - await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); - } - - Buffer.BlockCopy(WebSocketBuffer, 0, buffer, 0, buffer.Length); - WebSocketBufferSize -= buffer.Length; - WebSocketBufferOffset += buffer.Length; - } - else - { - Buffer.BlockCopy(WebSocketBuffer, WebSocketBufferOffset, buffer, 0, buffer.Length); - WebSocketBufferSize -= buffer.Length; - WebSocketBufferOffset += buffer.Length; - } - return; } + + var offset = 0; + while (_webSocket.State == WebSocketState.Open && WebSocketBufferSize < length) + { + WebSocketReceiveResult response; + do + { + response = await _webSocket.ReceiveAsync(new ArraySegment(buffer, offset, buffer.Length - offset), CancellationToken.None).ConfigureAwait(false); + offset += response.Count; + } while (!response.EndOfMessage); + + WebSocketBufferSize = response.Count; + if (response.MessageType == WebSocketMessageType.Close) + { + await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); + } + } } public Task WriteAsync(byte[] buffer) @@ -105,5 +91,10 @@ namespace MQTTnet.Implementations throw new MqttCommunicationException(exception); } } + + public int Peek() + { + return WebSocketBufferSize; + } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs b/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs index 8075f92d..b3e0080a 100644 --- a/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs +++ b/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs @@ -22,7 +22,7 @@ namespace MQTTnet { case MqttConnectionType.Tcp: case MqttConnectionType.Tls: - return new MqttTcpChannel(); + return new BufferedCommunicationChannel( new MqttTcpChannel() ); case MqttConnectionType.Ws: case MqttConnectionType.Wss: return new MqttWebSocketChannel(); diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index 5b20cf19..87c89b1d 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -92,17 +92,15 @@ namespace MQTTnet.Implementations } } - public async Task ReadAsync(byte[] buffer) + public async Task> ReadAsync(int length, byte[] buffer) { - if (buffer == null) throw new ArgumentNullException(nameof(buffer)); - try { var totalBytes = 0; do { - var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes).ConfigureAwait(false); + var read = await _dataStream.ReadAsync(buffer, totalBytes, length - totalBytes).ConfigureAwait(false); if (read == 0) { throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting)); @@ -110,7 +108,8 @@ namespace MQTTnet.Implementations totalBytes += read; } - while (totalBytes < buffer.Length); + while (totalBytes < length); + return new ArraySegment(buffer, 0, length); } catch (SocketException exception) { @@ -142,5 +141,10 @@ namespace MQTTnet.Implementations return certificates; } + + public int Peek() + { + return _socket.Available; + } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs index b90b02a8..2aa31b4a 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs @@ -11,9 +11,6 @@ namespace MQTTnet.Implementations public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable { private ClientWebSocket _webSocket = new ClientWebSocket(); - private const int BufferSize = 4096; - private const int BufferAmplifier = 20; - private readonly byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier]; private int WebSocketBufferSize; private int WebSocketBufferOffset; @@ -42,51 +39,40 @@ namespace MQTTnet.Implementations _webSocket?.Dispose(); } - public Task ReadAsync(byte[] buffer) + public async Task> ReadAsync(int length, byte[] buffer) { - return Task.WhenAll(ReadToBufferAsync(buffer)); + await ReadToBufferAsync(length, buffer).ConfigureAwait(false); + + var result = new ArraySegment(buffer, WebSocketBufferOffset, length); + WebSocketBufferSize -= length; + WebSocketBufferOffset += length; + + return result; } - private async Task ReadToBufferAsync(byte[] buffer) + private async Task ReadToBufferAsync(int length, byte[] buffer) { - var temporaryBuffer = new byte[BufferSize]; - var offset = 0; - - while (_webSocket.State == WebSocketState.Open) + if (WebSocketBufferSize > 0) { - if (WebSocketBufferSize == 0) - { - WebSocketBufferOffset = 0; - - WebSocketReceiveResult response; - do - { - response = await _webSocket.ReceiveAsync(new ArraySegment(temporaryBuffer), CancellationToken.None).ConfigureAwait(false); - - temporaryBuffer.CopyTo(WebSocketBuffer, offset); - offset += response.Count; - temporaryBuffer = new byte[BufferSize]; - } while (!response.EndOfMessage); - - WebSocketBufferSize = response.Count; - if (response.MessageType == WebSocketMessageType.Close) - { - await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); - } - - Buffer.BlockCopy(WebSocketBuffer, 0, buffer, 0, buffer.Length); - WebSocketBufferSize -= buffer.Length; - WebSocketBufferOffset += buffer.Length; - } - else - { - Buffer.BlockCopy(WebSocketBuffer, WebSocketBufferOffset, buffer, 0, buffer.Length); - WebSocketBufferSize -= buffer.Length; - WebSocketBufferOffset += buffer.Length; - } - return; } + + var offset = 0; + while (_webSocket.State == WebSocketState.Open && WebSocketBufferSize < length) + { + WebSocketReceiveResult response; + do + { + response = await _webSocket.ReceiveAsync(new ArraySegment(buffer, offset, buffer.Length - offset), CancellationToken.None).ConfigureAwait(false); + offset += response.Count; + } while (!response.EndOfMessage); + + WebSocketBufferSize = response.Count; + if (response.MessageType == WebSocketMessageType.Close) + { + await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false); + } + } } public Task WriteAsync(byte[] buffer) @@ -105,5 +91,10 @@ namespace MQTTnet.Implementations throw new MqttCommunicationException(exception); } } + + public int Peek() + { + return WebSocketBufferSize; + } } } \ No newline at end of file diff --git a/MQTTnet.Core/Channel/BufferConstants.cs b/MQTTnet.Core/Channel/BufferConstants.cs new file mode 100644 index 00000000..3b441e0c --- /dev/null +++ b/MQTTnet.Core/Channel/BufferConstants.cs @@ -0,0 +1,7 @@ +namespace MQTTnet.Core.Channel +{ + public static class BufferConstants + { + public const int Size = 4096 * 20; + } +} diff --git a/MQTTnet.Core/Channel/BufferedCommunicationChannel.cs b/MQTTnet.Core/Channel/BufferedCommunicationChannel.cs new file mode 100644 index 00000000..34ee586f --- /dev/null +++ b/MQTTnet.Core/Channel/BufferedCommunicationChannel.cs @@ -0,0 +1,77 @@ +using System.Threading.Tasks; +using MQTTnet.Core.Client; +using System; + +namespace MQTTnet.Core.Channel +{ + public class BufferedCommunicationChannel : IMqttCommunicationChannel + { + private IMqttCommunicationChannel _inner { get; } + private int _bufferSize = 0; + private int _bufferOffset = 0; + + public BufferedCommunicationChannel(IMqttCommunicationChannel inner) + { + _inner = inner; + } + + public Task ConnectAsync(MqttClientOptions options) + { + return _inner.ConnectAsync(options); + } + + public Task DisconnectAsync() + { + return _inner.DisconnectAsync(); + } + + public int Peek() + { + return _inner.Peek(); + } + + public async Task> ReadAsync(int length, byte[] buffer) + { + //read from buffer + if (_bufferSize > 0) + { + return ReadFomBuffer(length, buffer); + } + + var available = _inner.Peek(); + // if there are less or equal bytes available then requested then just read em + if (available <= length) + { + return await _inner.ReadAsync(length, buffer); + } + + //if more bytes are available than requested do buffer them to reduce calls to network buffers + await WriteToBuffer(available, buffer).ConfigureAwait(false); + return ReadFomBuffer(length, buffer); + } + + private async Task WriteToBuffer(int available, byte[] buffer) + { + await _inner.ReadAsync(available, buffer).ConfigureAwait(false); + _bufferSize = available; + _bufferOffset = 0; + } + + private ArraySegment ReadFomBuffer(int length, byte[] buffer) + { + var result = new ArraySegment(buffer, _bufferOffset, length); + _bufferSize -= length; + _bufferOffset += length; + + if (_bufferSize < 0) + { + } + return result; + } + + public Task WriteAsync(byte[] buffer) + { + return _inner.WriteAsync(buffer); + } + } +} diff --git a/MQTTnet.Core/Channel/IMqttCommunicationChannel.cs b/MQTTnet.Core/Channel/IMqttCommunicationChannel.cs index a1ec890e..0f6ea4b6 100644 --- a/MQTTnet.Core/Channel/IMqttCommunicationChannel.cs +++ b/MQTTnet.Core/Channel/IMqttCommunicationChannel.cs @@ -1,5 +1,6 @@ using System.Threading.Tasks; using MQTTnet.Core.Client; +using System; namespace MQTTnet.Core.Channel { @@ -11,6 +12,11 @@ namespace MQTTnet.Core.Channel Task WriteAsync(byte[] buffer); - Task ReadAsync(byte[] buffer); + /// + /// get the currently available number of bytes without reading them + /// + int Peek(); + + Task> ReadAsync(int length, byte[] buffer); } } diff --git a/MQTTnet.Core/Packets/MqttPacketHeader.cs b/MQTTnet.Core/Packets/MqttPacketHeader.cs new file mode 100644 index 00000000..41646df4 --- /dev/null +++ b/MQTTnet.Core/Packets/MqttPacketHeader.cs @@ -0,0 +1,13 @@ +using MQTTnet.Core.Protocol; + +namespace MQTTnet.Core.Packets +{ + public class MqttPacketHeader + { + public MqttControlPacketType ControlPacketType { get; set; } + + public byte FixedHeader { get; set; } + + public int BodyLength { get; set; } + } +} diff --git a/MQTTnet.Core/Serializer/MqttPacketReader.cs b/MQTTnet.Core/Serializer/MqttPacketReader.cs index f64c0c9c..38d3c7c3 100644 --- a/MQTTnet.Core/Serializer/MqttPacketReader.cs +++ b/MQTTnet.Core/Serializer/MqttPacketReader.cs @@ -2,55 +2,28 @@ using System.IO; using System.Text; using System.Threading.Tasks; -using MQTTnet.Core.Channel; using MQTTnet.Core.Exceptions; using MQTTnet.Core.Protocol; +using MQTTnet.Core.Channel; +using MQTTnet.Core.Packets; namespace MQTTnet.Core.Serializer { - public sealed class MqttPacketReader : IDisposable + public sealed class MqttPacketReader : BinaryReader { - private readonly MemoryStream _remainingData = new MemoryStream(1024); - private readonly IMqttCommunicationChannel _source; + private readonly MqttPacketHeader _header; - private int _remainingLength; - - public MqttPacketReader(IMqttCommunicationChannel source) + public MqttPacketReader(Stream stream, MqttPacketHeader header) + : base(stream) { - _source = source ?? throw new ArgumentNullException(nameof(source)); + _header = header; } - - public MqttControlPacketType ControlPacketType { get; private set; } - - public byte FixedHeader { get; private set; } - - public bool EndOfRemainingData => _remainingData.Position == _remainingData.Length; - - public async Task ReadToEndAsync() + + public bool EndOfRemainingData => BaseStream.Position == _header.BodyLength; + + public override ushort ReadUInt16() { - await ReadFixedHeaderAsync().ConfigureAwait(false); - await ReadRemainingLengthAsync().ConfigureAwait(false); - - if (_remainingLength == 0) - { - return; - } - - var buffer = new byte[_remainingLength]; - await ReadFromSourceAsync(buffer).ConfigureAwait(false); - - _remainingData.Write(buffer, 0, buffer.Length); - _remainingData.Position = 0; - } - - public byte ReadRemainingDataByte() - { - return ReadRemainingData(1)[0]; - } - - public ushort ReadRemainingDataUShort() - { - var buffer = ReadRemainingData(2); + var buffer = ReadBytes(2); var temp = buffer[0]; buffer[0] = buffer[1]; @@ -59,31 +32,58 @@ namespace MQTTnet.Core.Serializer return BitConverter.ToUInt16(buffer, 0); } - public string ReadRemainingDataStringWithLengthPrefix() + public string ReadStringWithLengthPrefix() { - var buffer = ReadRemainingDataWithLengthPrefix(); + var buffer = ReadWithLengthPrefix(); return Encoding.UTF8.GetString(buffer, 0, buffer.Length); } - public byte[] ReadRemainingDataWithLengthPrefix() + public byte[] ReadWithLengthPrefix() { - var length = ReadRemainingDataUShort(); - return ReadRemainingData(length); + var length = ReadUInt16(); + return ReadBytes(length); } public byte[] ReadRemainingData() { - return ReadRemainingData(_remainingLength - (int)_remainingData.Position); + return ReadBytes(_header.BodyLength - (int)BaseStream.Position); } - public byte[] ReadRemainingData(int length) + public static async Task ReadHeaderFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer) { - var buffer = new byte[length]; - _remainingData.Read(buffer, 0, buffer.Length); - return buffer; + var fixedHeader = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false); + var byteReader = new ByteReader(fixedHeader); + byteReader.Read(4); + var controlPacketType = (MqttControlPacketType)byteReader.Read(4); + var bodyLength = await ReadBodyLengthFromSourceAsync(source, buffer).ConfigureAwait(false); + + return new MqttPacketHeader() + { + FixedHeader = fixedHeader, + ControlPacketType = controlPacketType, + BodyLength = bodyLength + }; } - private async Task ReadRemainingLengthAsync() + private static async Task ReadStreamByteAsync(IMqttCommunicationChannel source, byte[] readBuffer) + { + var result = await ReadFromSourceAsync(source, 1, readBuffer).ConfigureAwait(false); + return result.Array[result.Offset]; + } + + public static async Task> ReadFromSourceAsync(IMqttCommunicationChannel source, int length, byte[] buffer) + { + try + { + return await source.ReadAsync(length, buffer); + } + catch (Exception exception) + { + throw new MqttCommunicationException(exception); + } + } + + private static async Task ReadBodyLengthFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer) { // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html. var multiplier = 1; @@ -91,7 +91,7 @@ namespace MQTTnet.Core.Serializer byte encodedByte; do { - encodedByte = await ReadStreamByteAsync().ConfigureAwait(false); + encodedByte = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false); value += (encodedByte & 127) * multiplier; multiplier *= 128; if (multiplier > 128 * 128 * 128) @@ -99,41 +99,7 @@ namespace MQTTnet.Core.Serializer throw new MqttProtocolViolationException("Remaining length is ivalid."); } } while ((encodedByte & 128) != 0); - - _remainingLength = value; - } - - private Task ReadFromSourceAsync(byte[] buffer) - { - try - { - return _source.ReadAsync(buffer); - } - catch (Exception exception) - { - throw new MqttCommunicationException(exception); - } - } - - private async Task ReadStreamByteAsync() - { - var buffer = new byte[1]; - await ReadFromSourceAsync(buffer).ConfigureAwait(false); - return buffer[0]; - } - - private async Task ReadFixedHeaderAsync() - { - FixedHeader = await ReadStreamByteAsync().ConfigureAwait(false); - - var byteReader = new ByteReader(FixedHeader); - byteReader.Read(4); - ControlPacketType = (MqttControlPacketType)byteReader.Read(4); - } - - public void Dispose() - { - _remainingData?.Dispose(); + return value; } } } diff --git a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs b/MQTTnet.Core/Serializer/MqttPacketSerializer.cs index e2e8f447..064b79b6 100644 --- a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs +++ b/MQTTnet.Core/Serializer/MqttPacketSerializer.cs @@ -1,4 +1,6 @@ using System; +using System.Collections.Generic; +using System.IO; using System.Linq; using System.Text; using System.Threading.Tasks; @@ -15,80 +17,97 @@ namespace MQTTnet.Core.Serializer private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs"); public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; + private byte[] _readBuffer = new byte[BufferConstants.Size]; - public Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination) + public async Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination) { 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)) + { + var header = new List(); + header.Add(SerializePacket(packet, writer)); + + var body = stream.ToArray(); + MqttPacketWriter.BuildLengthHeader(body.Length, header); + + await destination.WriteAsync(header.ToArray()).ConfigureAwait(false); + await destination.WriteAsync(body).ConfigureAwait(false); + } + } + + private byte SerializePacket(MqttBasePacket packet, MqttPacketWriter writer) + { if (packet is MqttConnectPacket connectPacket) { - return SerializeAsync(connectPacket, destination); + return Serialize(connectPacket, writer); } if (packet is MqttConnAckPacket connAckPacket) { - return SerializeAsync(connAckPacket, destination); + return Serialize(connAckPacket, writer); } if (packet is MqttDisconnectPacket disconnectPacket) { - return SerializeAsync(disconnectPacket, destination); + return Serialize(disconnectPacket, writer); } if (packet is MqttPingReqPacket pingReqPacket) { - return SerializeAsync(pingReqPacket, destination); + return Serialize(pingReqPacket, writer); } if (packet is MqttPingRespPacket pingRespPacket) { - return SerializeAsync(pingRespPacket, destination); + return Serialize(pingRespPacket, writer); } if (packet is MqttPublishPacket publishPacket) { - return SerializeAsync(publishPacket, destination); + return Serialize(publishPacket, writer); } if (packet is MqttPubAckPacket pubAckPacket) { - return SerializeAsync(pubAckPacket, destination); + return Serialize(pubAckPacket, writer); } if (packet is MqttPubRecPacket pubRecPacket) { - return SerializeAsync(pubRecPacket, destination); + return Serialize(pubRecPacket, writer); } if (packet is MqttPubRelPacket pubRelPacket) { - return SerializeAsync(pubRelPacket, destination); + return Serialize(pubRelPacket, writer); } if (packet is MqttPubCompPacket pubCompPacket) { - return SerializeAsync(pubCompPacket, destination); + return Serialize(pubCompPacket, writer); } if (packet is MqttSubscribePacket subscribePacket) { - return SerializeAsync(subscribePacket, destination); + return Serialize(subscribePacket, writer); } if (packet is MqttSubAckPacket subAckPacket) { - return SerializeAsync(subAckPacket, destination); + return Serialize(subAckPacket, writer); } if (packet is MqttUnsubscribePacket unsubscribePacket) { - return SerializeAsync(unsubscribePacket, destination); + return Serialize(unsubscribePacket, writer); } if (packet is MqttUnsubAckPacket unsubAckPacket) { - return SerializeAsync(unsubAckPacket, destination); + return Serialize(unsubAckPacket, writer); } throw new MqttProtocolViolationException("Packet type invalid."); @@ -98,11 +117,13 @@ namespace MQTTnet.Core.Serializer { if (source == null) throw new ArgumentNullException(nameof(source)); - using (var mqttPacketReader = new MqttPacketReader(source)) - { - await mqttPacketReader.ReadToEndAsync().ConfigureAwait(false); + var header = await MqttPacketReader.ReadHeaderFromSourceAsync(source, _readBuffer).ConfigureAwait(false); - switch (mqttPacketReader.ControlPacketType) + var body = await GetBody(source, header).ConfigureAwait(false); + + using (var mqttPacketReader = new MqttPacketReader(body, header)) + { + switch (header.ControlPacketType) { case MqttControlPacketType.Connect: { @@ -121,14 +142,14 @@ namespace MQTTnet.Core.Serializer case MqttControlPacketType.Publish: { - return DeserializePublish(mqttPacketReader); + return DeserializePublish(mqttPacketReader, header); } case MqttControlPacketType.PubAck: { return new MqttPubAckPacket { - PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() + PacketIdentifier = mqttPacketReader.ReadUInt16() }; } @@ -136,7 +157,7 @@ namespace MQTTnet.Core.Serializer { return new MqttPubRecPacket { - PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() + PacketIdentifier = mqttPacketReader.ReadUInt16() }; } @@ -144,7 +165,7 @@ namespace MQTTnet.Core.Serializer { return new MqttPubRelPacket { - PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() + PacketIdentifier = mqttPacketReader.ReadUInt16() }; } @@ -152,7 +173,7 @@ namespace MQTTnet.Core.Serializer { return new MqttPubCompPacket { - PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() + PacketIdentifier = mqttPacketReader.ReadUInt16() }; } @@ -185,28 +206,39 @@ namespace MQTTnet.Core.Serializer { return new MqttUnsubAckPacket { - PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() + PacketIdentifier = mqttPacketReader.ReadUInt16() }; } default: { - throw new MqttProtocolViolationException($"Packet type ({(int)mqttPacketReader.ControlPacketType}) not supported."); + throw new MqttProtocolViolationException($"Packet type ({(int)header.ControlPacketType}) not supported."); } } } } + private async Task GetBody(IMqttCommunicationChannel source, MqttPacketHeader header) + { + if (header.BodyLength > 0) + { + var segment = await MqttPacketReader.ReadFromSourceAsync(source, header.BodyLength, _readBuffer).ConfigureAwait(false); + return new MemoryStream(segment.Array, segment.Offset, segment.Count); + } + + return new MemoryStream(); + } + private static MqttBasePacket DeserializeUnsubscribe(MqttPacketReader reader) { var packet = new MqttUnsubscribePacket { - PacketIdentifier = reader.ReadRemainingDataUShort(), + PacketIdentifier = reader.ReadUInt16(), }; while (!reader.EndOfRemainingData) { - packet.TopicFilters.Add(reader.ReadRemainingDataStringWithLengthPrefix()); + packet.TopicFilters.Add(reader.ReadStringWithLengthPrefix()); } return packet; @@ -216,32 +248,32 @@ namespace MQTTnet.Core.Serializer { var packet = new MqttSubscribePacket { - PacketIdentifier = reader.ReadRemainingDataUShort() + PacketIdentifier = reader.ReadUInt16() }; while (!reader.EndOfRemainingData) { packet.TopicFilters.Add(new TopicFilter( - reader.ReadRemainingDataStringWithLengthPrefix(), - (MqttQualityOfServiceLevel)reader.ReadRemainingDataByte())); + reader.ReadStringWithLengthPrefix(), + (MqttQualityOfServiceLevel)reader.ReadByte())); } return packet; } - private static MqttBasePacket DeserializePublish(MqttPacketReader reader) + private static MqttBasePacket DeserializePublish(MqttPacketReader reader, MqttPacketHeader mqttPacketHeader) { - var fixedHeader = new ByteReader(reader.FixedHeader); + var fixedHeader = new ByteReader(mqttPacketHeader.FixedHeader); var retain = fixedHeader.Read(); var qualityOfServiceLevel = (MqttQualityOfServiceLevel)fixedHeader.Read(2); var dup = fixedHeader.Read(); - var topic = reader.ReadRemainingDataStringWithLengthPrefix(); + var topic = reader.ReadStringWithLengthPrefix(); ushort packetIdentifier = 0; if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - packetIdentifier = reader.ReadRemainingDataUShort(); + packetIdentifier = reader.ReadUInt16(); } var packet = new MqttPublishPacket @@ -259,13 +291,13 @@ namespace MQTTnet.Core.Serializer private static MqttBasePacket DeserializeConnect(MqttPacketReader reader) { - reader.ReadRemainingData(2); // Skip 2 bytes + reader.ReadBytes(2); // Skip 2 bytes MqttProtocolVersion protocolVersion; - var protocolName = reader.ReadRemainingData(4); + var protocolName = reader.ReadBytes(4); if (protocolName.SequenceEqual(ProtocolVersionV310Name)) { - reader.ReadRemainingData(2); + reader.ReadBytes(2); protocolVersion = MqttProtocolVersion.V310; } else if (protocolName.SequenceEqual(ProtocolVersionV311Name)) @@ -277,8 +309,8 @@ namespace MQTTnet.Core.Serializer throw new MqttProtocolViolationException("Protocol name is not supported."); } - var protocolLevel = reader.ReadRemainingDataByte(); - var connectFlags = reader.ReadRemainingDataByte(); + var protocolLevel = reader.ReadByte(); + var connectFlags = reader.ReadByte(); var connectFlagsReader = new ByteReader(connectFlags); connectFlagsReader.Read(); // Reserved. @@ -295,26 +327,26 @@ namespace MQTTnet.Core.Serializer var passwordFlag = connectFlagsReader.Read(); var usernameFlag = connectFlagsReader.Read(); - packet.KeepAlivePeriod = reader.ReadRemainingDataUShort(); - packet.ClientId = reader.ReadRemainingDataStringWithLengthPrefix(); + packet.KeepAlivePeriod = reader.ReadUInt16(); + packet.ClientId = reader.ReadStringWithLengthPrefix(); if (willFlag) { packet.WillMessage = new MqttApplicationMessage( - reader.ReadRemainingDataStringWithLengthPrefix(), - reader.ReadRemainingDataWithLengthPrefix(), + reader.ReadStringWithLengthPrefix(), + reader.ReadWithLengthPrefix(), (MqttQualityOfServiceLevel)willQoS, willRetain); } if (usernameFlag) { - packet.Username = reader.ReadRemainingDataStringWithLengthPrefix(); + packet.Username = reader.ReadStringWithLengthPrefix(); } if (passwordFlag) { - packet.Password = reader.ReadRemainingDataStringWithLengthPrefix(); + packet.Password = reader.ReadStringWithLengthPrefix(); } ValidateConnectPacket(packet); @@ -325,12 +357,12 @@ namespace MQTTnet.Core.Serializer { var packet = new MqttSubAckPacket { - PacketIdentifier = reader.ReadRemainingDataUShort() + PacketIdentifier = reader.ReadUInt16() }; while (!reader.EndOfRemainingData) { - packet.SubscribeReturnCodes.Add((MqttSubscribeReturnCode)reader.ReadRemainingDataByte()); + packet.SubscribeReturnCodes.Add((MqttSubscribeReturnCode)reader.ReadByte()); } return packet; @@ -338,8 +370,8 @@ namespace MQTTnet.Core.Serializer private static MqttBasePacket DeserializeConnAck(MqttPacketReader reader) { - var variableHeader1 = reader.ReadRemainingDataByte(); - var variableHeader2 = reader.ReadRemainingDataByte(); + var variableHeader1 = reader.ReadByte(); + var variableHeader2 = reader.ReadByte(); var packet = new MqttConnAckPacket { @@ -366,260 +398,212 @@ namespace MQTTnet.Core.Serializer } } - private Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination) + private byte Serialize(MqttConnectPacket packet, MqttPacketWriter writer) { ValidateConnectPacket(packet); - using (var output = new MqttPacketWriter()) + // Write variable header + writer.Write(0x00, 0x04); // 3.1.2.1 Protocol Name + if (ProtocolVersion == MqttProtocolVersion.V311) { - // Write variable header - output.Write(0x00, 0x04); // 3.1.2.1 Protocol Name - if (ProtocolVersion == MqttProtocolVersion.V311) - { - output.Write(ProtocolVersionV311Name); - output.Write(0x04); // 3.1.2.2 Protocol Level (4) - } - else - { - output.Write(ProtocolVersionV310Name); - output.Write(0x64); - output.Write(0x70); - output.Write(0x03); // Protocol Level (3) - } - - var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags - connectFlags.Write(false); // Reserved - connectFlags.Write(packet.CleanSession); - connectFlags.Write(packet.WillMessage != null); - - if (packet.WillMessage != null) - { - connectFlags.Write((int)packet.WillMessage.QualityOfServiceLevel, 2); - connectFlags.Write(packet.WillMessage.Retain); - } - else - { - connectFlags.Write(0, 2); - connectFlags.Write(false); - } - - connectFlags.Write(packet.Password != null); - connectFlags.Write(packet.Username != null); - - output.Write(connectFlags); - output.Write(packet.KeepAlivePeriod); - output.WriteWithLengthPrefix(packet.ClientId); - - if (packet.WillMessage != null) - { - output.WriteWithLengthPrefix(packet.WillMessage.Topic); - output.WriteWithLengthPrefix(packet.WillMessage.Payload); - } - - if (packet.Username != null) - { - output.WriteWithLengthPrefix(packet.Username); - } - - if (packet.Password != null) - { - output.WriteWithLengthPrefix(packet.Password); - } - - output.InjectFixedHeader(MqttControlPacketType.Connect); - return output.WriteToAsync(destination); + writer.Write(ProtocolVersionV311Name); + writer.Write(0x04); // 3.1.2.2 Protocol Level (4) } - } - - private Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination) - { - using (var output = new MqttPacketWriter()) + else { - var connectAcknowledgeFlags = new ByteWriter(); - - if (ProtocolVersion == MqttProtocolVersion.V311) - { - connectAcknowledgeFlags.Write(packet.IsSessionPresent); - } - - output.Write(connectAcknowledgeFlags); - output.Write((byte)packet.ConnectReturnCode); - - output.InjectFixedHeader(MqttControlPacketType.ConnAck); - return output.WriteToAsync(destination); + writer.Write(ProtocolVersionV310Name); + writer.Write(0x64); + writer.Write(0x70); + writer.Write(0x03); // Protocol Level (3) } - } - private static async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination) - { - using (var output = new MqttPacketWriter()) + var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags + connectFlags.Write(false); // Reserved + connectFlags.Write(packet.CleanSession); + connectFlags.Write(packet.WillMessage != null); + + if (packet.WillMessage != null) { - output.Write(packet.PacketIdentifier); - - output.InjectFixedHeader(MqttControlPacketType.PubRel, 0x02); - await output.WriteToAsync(destination).ConfigureAwait(false); + connectFlags.Write((int)packet.WillMessage.QualityOfServiceLevel, 2); + connectFlags.Write(packet.WillMessage.Retain); } + else + { + connectFlags.Write(0, 2); + connectFlags.Write(false); + } + + connectFlags.Write(packet.Password != null); + connectFlags.Write(packet.Username != null); + + writer.Write(connectFlags); + writer.Write(packet.KeepAlivePeriod); + writer.WriteWithLengthPrefix(packet.ClientId); + + if (packet.WillMessage != null) + { + writer.WriteWithLengthPrefix(packet.WillMessage.Topic); + writer.WriteWithLengthPrefix(packet.WillMessage.Payload); + } + + if (packet.Username != null) + { + writer.WriteWithLengthPrefix(packet.Username); + } + + if (packet.Password != null) + { + writer.WriteWithLengthPrefix(packet.Password); + } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); } - private static Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination) + private byte Serialize(MqttConnAckPacket packet, MqttPacketWriter writer) { - return SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, destination); + var connectAcknowledgeFlags = new ByteWriter(); + + if (ProtocolVersion == MqttProtocolVersion.V311) + { + connectAcknowledgeFlags.Write(packet.IsSessionPresent); + } + + writer.Write(connectAcknowledgeFlags); + writer.Write((byte)packet.ConnectReturnCode); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); } - private static Task SerializeAsync(MqttPingReqPacket packet, IMqttCommunicationChannel destination) + private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter writer) { - return SerializeEmptyPacketAsync(MqttControlPacketType.PingReq, destination); + writer.Write(packet.PacketIdentifier); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); } - private static Task SerializeAsync(MqttPingRespPacket packet, IMqttCommunicationChannel destination) + private static byte Serialize(MqttDisconnectPacket packet, MqttPacketWriter writer) { - return SerializeEmptyPacketAsync(MqttControlPacketType.PingResp, destination); + return SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, writer); } - private static Task SerializeAsync(MqttPublishPacket packet, IMqttCommunicationChannel destination) + private static byte Serialize(MqttPingReqPacket packet, MqttPacketWriter writer) + { + return SerializeEmptyPacketAsync(MqttControlPacketType.PingReq, writer); + } + + private static byte Serialize(MqttPingRespPacket packet, MqttPacketWriter writer) + { + return SerializeEmptyPacketAsync(MqttControlPacketType.PingResp, writer); + } + + private static byte Serialize(MqttPublishPacket packet, MqttPacketWriter writer) { ValidatePublishPacket(packet); - using (var output = new MqttPacketWriter()) - { - output.WriteWithLengthPrefix(packet.Topic); + writer.WriteWithLengthPrefix(packet.Topic); - if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + { + writer.Write(packet.PacketIdentifier); + } + else + { + if (packet.PacketIdentifier > 0) { - output.Write(packet.PacketIdentifier); + throw new MqttProtocolViolationException("Packet identifier must be empty if QoS == 0 [MQTT-2.3.1-5]."); } - else + } + + if (packet.Payload?.Length > 0) + { + writer.Write(packet.Payload); + } + + var fixedHeader = new ByteWriter(); + fixedHeader.Write(packet.Retain); + fixedHeader.Write((byte)packet.QualityOfServiceLevel, 2); + fixedHeader.Write(packet.Dup); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader.Value); + } + + private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer) + { + writer.Write(packet.PacketIdentifier); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck); + } + + private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter writer) + { + writer.Write(packet.PacketIdentifier); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec); + } + + private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter writer) + { + writer.Write(packet.PacketIdentifier); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp); + } + + private static byte Serialize(MqttSubscribePacket packet, MqttPacketWriter writer) + { + writer.Write(packet.PacketIdentifier); + + if (packet.TopicFilters?.Count > 0) + { + foreach (var topicFilter in packet.TopicFilters) { - if (packet.PacketIdentifier > 0) - { - throw new MqttProtocolViolationException("Packet identifier must be empty if QoS == 0 [MQTT-2.3.1-5]."); - } + writer.WriteWithLengthPrefix(topicFilter.Topic); + writer.Write((byte)topicFilter.QualityOfServiceLevel); } + } - if (packet.Payload?.Length > 0) + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02); + } + + private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter writer) + { + writer.Write(packet.PacketIdentifier); + + if (packet.SubscribeReturnCodes?.Any() == true) + { + foreach (var packetSubscribeReturnCode in packet.SubscribeReturnCodes) { - output.Write(packet.Payload); + writer.Write((byte)packetSubscribeReturnCode); } - - var fixedHeader = new ByteWriter(); - fixedHeader.Write(packet.Retain); - fixedHeader.Write((byte)packet.QualityOfServiceLevel, 2); - fixedHeader.Write(packet.Dup); - - output.InjectFixedHeader(MqttControlPacketType.Publish, fixedHeader.Value); - return output.WriteToAsync(destination); } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck); } - private static Task SerializeAsync(MqttPubAckPacket packet, IMqttCommunicationChannel destination) + private static byte Serialize(MqttUnsubscribePacket packet, MqttPacketWriter writer) { - using (var output = new MqttPacketWriter()) + writer.Write(packet.PacketIdentifier); + + if (packet.TopicFilters?.Any() == true) { - output.Write(packet.PacketIdentifier); - - output.InjectFixedHeader(MqttControlPacketType.PubAck); - return output.WriteToAsync(destination); - } - } - - private static Task SerializeAsync(MqttPubRecPacket packet, IMqttCommunicationChannel destination) - { - using (var output = new MqttPacketWriter()) - { - output.Write(packet.PacketIdentifier); - - output.InjectFixedHeader(MqttControlPacketType.PubRec); - return output.WriteToAsync(destination); - } - } - - private static Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination) - { - using (var output = new MqttPacketWriter()) - { - output.Write(packet.PacketIdentifier); - - output.InjectFixedHeader(MqttControlPacketType.PubComp); - return output.WriteToAsync(destination); - } - } - - private static Task SerializeAsync(MqttSubscribePacket packet, IMqttCommunicationChannel destination) - { - using (var output = new MqttPacketWriter()) - { - output.Write(packet.PacketIdentifier); - - if (packet.TopicFilters?.Count > 0) + foreach (var topicFilter in packet.TopicFilters) { - foreach (var topicFilter in packet.TopicFilters) - { - output.WriteWithLengthPrefix(topicFilter.Topic); - output.Write((byte)topicFilter.QualityOfServiceLevel); - } + writer.WriteWithLengthPrefix(topicFilter); } - - output.InjectFixedHeader(MqttControlPacketType.Subscribe, 0x02); - return output.WriteToAsync(destination); } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02); } - private static Task SerializeAsync(MqttSubAckPacket packet, IMqttCommunicationChannel destination) + private static byte Serialize(MqttUnsubAckPacket packet, MqttPacketWriter writer) { - using (var output = new MqttPacketWriter()) - { - output.Write(packet.PacketIdentifier); + writer.Write(packet.PacketIdentifier); - if (packet.SubscribeReturnCodes?.Any() == true) - { - foreach (var packetSubscribeReturnCode in packet.SubscribeReturnCodes) - { - output.Write((byte)packetSubscribeReturnCode); - } - } - - output.InjectFixedHeader(MqttControlPacketType.SubAck); - return output.WriteToAsync(destination); - } + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck); } - private static Task SerializeAsync(MqttUnsubscribePacket packet, IMqttCommunicationChannel destination) + private static byte SerializeEmptyPacketAsync(MqttControlPacketType type, MqttPacketWriter writer) { - using (var output = new MqttPacketWriter()) - { - output.Write(packet.PacketIdentifier); - - if (packet.TopicFilters?.Any() == true) - { - foreach (var topicFilter in packet.TopicFilters) - { - output.WriteWithLengthPrefix(topicFilter); - } - } - - output.InjectFixedHeader(MqttControlPacketType.Unsubscibe, 0x02); - return output.WriteToAsync(destination); - } - } - - private static Task SerializeAsync(MqttUnsubAckPacket packet, IMqttCommunicationChannel destination) - { - using (var output = new MqttPacketWriter()) - { - output.Write(packet.PacketIdentifier); - - output.InjectFixedHeader(MqttControlPacketType.UnsubAck); - return output.WriteToAsync(destination); - } - } - - private static Task SerializeEmptyPacketAsync(MqttControlPacketType type, IMqttCommunicationChannel destination) - { - using (var output = new MqttPacketWriter()) - { - output.InjectFixedHeader(type); - return output.WriteToAsync(destination); - } + return MqttPacketWriter.BuildFixedHeader(type); } } } diff --git a/MQTTnet.Core/Serializer/MqttPacketWriter.cs b/MQTTnet.Core/Serializer/MqttPacketWriter.cs index 778f52e7..d5aa47a0 100644 --- a/MQTTnet.Core/Serializer/MqttPacketWriter.cs +++ b/MQTTnet.Core/Serializer/MqttPacketWriter.cs @@ -1,49 +1,45 @@ using System; using System.IO; using System.Text; -using System.Threading.Tasks; -using MQTTnet.Core.Channel; using MQTTnet.Core.Protocol; +using System.Collections.Generic; namespace MQTTnet.Core.Serializer { - public sealed class MqttPacketWriter : IDisposable + public sealed class MqttPacketWriter : BinaryWriter { - private readonly MemoryStream _buffer = new MemoryStream(1024); + public MqttPacketWriter( Stream stream ) + : base(stream) + { - public void InjectFixedHeader(MqttControlPacketType packetType, byte flags = 0) + } + + public static byte BuildFixedHeader(MqttControlPacketType packetType, byte flags = 0) { var fixedHeader = (int)packetType << 4; fixedHeader |= flags; - InjectFixedHeader((byte)fixedHeader); + return (byte)fixedHeader; } - - public void Write(byte value) - { - _buffer.WriteByte(value); - } - - public void Write(ushort value) + + public override void Write(ushort value) { var buffer = BitConverter.GetBytes(value); - _buffer.WriteByte(buffer[1]); - _buffer.WriteByte(buffer[0]); + Write(buffer[1]); + Write(buffer[0]); + } + + public new void Write(params byte[] values) + { + base.Write(values); } public void Write(ByteWriter value) { if (value == null) throw new ArgumentNullException(nameof(value)); - _buffer.WriteByte(value.Value); + Write(value.Value); } - - public void Write(params byte[] value) - { - if (value == null) throw new ArgumentNullException(nameof(value)); - - _buffer.Write(value, 0, value.Length); - } - + public void WriteWithLengthPrefix(string value) { WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty)); @@ -57,36 +53,16 @@ namespace MQTTnet.Core.Serializer Write(value); } - public Task WriteToAsync(IMqttCommunicationChannel destination) + public static void BuildLengthHeader(int length, List header) { - if (destination == null) throw new ArgumentNullException(nameof(destination)); - - return destination.WriteAsync(_buffer.ToArray()); - } - - public void Dispose() - { - _buffer?.Dispose(); - } - - private void InjectFixedHeader(byte fixedHeader) - { - if (_buffer.Length == 0) + if (length == 0) { - Write(fixedHeader); - Write(0); + header.Add(0); return; } - var backupBuffer = _buffer.ToArray(); - var remainingLength = (int)_buffer.Length; - - _buffer.SetLength(0); - - _buffer.WriteByte(fixedHeader); - // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html. - var x = remainingLength; + var x = length; do { var encodedByte = x % 128; @@ -96,10 +72,8 @@ namespace MQTTnet.Core.Serializer encodedByte = encodedByte | 128; } - _buffer.WriteByte((byte)encodedByte); + header.Add((byte)encodedByte); } while (x > 0); - - _buffer.Write(backupBuffer, 0, backupBuffer.Length); } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index fad1f4e6..6a18f50f 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -418,15 +418,21 @@ namespace MQTTnet.Core.Tests return _stream.WriteAsync(buffer, 0, buffer.Length); } - public Task ReadAsync(byte[] buffer) + public async Task> ReadAsync(int length, byte[] buffer) { - return _stream.ReadAsync(buffer, 0, buffer.Length); + await _stream.ReadAsync(buffer, 0, length); + return new ArraySegment(buffer, 0, length); } public byte[] ToArray() { return _stream.ToArray(); } + + public int Peek() + { + return (int)_stream.Length - (int)_stream.Position; + } } private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) diff --git a/Tests/MQTTnet.TestApp.NetFramework/MQTTnet.TestApp.NetFramework.csproj b/Tests/MQTTnet.TestApp.NetFramework/MQTTnet.TestApp.NetFramework.csproj index 2bf571e9..7d112f5a 100644 --- a/Tests/MQTTnet.TestApp.NetFramework/MQTTnet.TestApp.NetFramework.csproj +++ b/Tests/MQTTnet.TestApp.NetFramework/MQTTnet.TestApp.NetFramework.csproj @@ -82,6 +82,7 @@ + diff --git a/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs new file mode 100644 index 00000000..32750d1b --- /dev/null +++ b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs @@ -0,0 +1,165 @@ +using MQTTnet.Core; +using MQTTnet.Core.Client; +using MQTTnet.Core.Packets; +using MQTTnet.Core.Protocol; +using MQTTnet.Core.Server; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace MQTTnet.TestApp.NetFramework +{ + public static class PerformanceTest + { + public static async Task RunAsync() + { + var server = Task.Run(() => RunServerAsync()); + var client = Task.Run(() => RunClientAsync(50, TimeSpan.FromMilliseconds(10))); + + await Task.WhenAll(server, client).ConfigureAwait(false); + } + + private static async Task RunClientAsync( int msgChunkSize, TimeSpan interval ) + { + try + { + var options = new MqttClientOptions + { + Server = "localhost", + ClientId = "XYZ", + CleanSession = true, + DefaultCommunicationTimeout = TimeSpan.FromMinutes(10) + }; + + var client = new MqttClientFactory().CreateMqttClient(options); + client.ApplicationMessageReceived += (s, e) => + { + }; + + client.Connected += async (s, e) => + { + Console.WriteLine("### CONNECTED WITH SERVER ###"); + + await client.SubscribeAsync(new List + { + new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce) + }); + + Console.WriteLine("### SUBSCRIBED ###"); + }; + + client.Disconnected += async (s, e) => + { + Console.WriteLine("### DISCONNECTED FROM SERVER ###"); + await Task.Delay(TimeSpan.FromSeconds(5)); + + try + { + await client.ConnectAsync(); + } + catch + { + Console.WriteLine("### RECONNECTING FAILED ###"); + } + }; + + try + { + await client.ConnectAsync(); + } + catch (Exception exception) + { + Console.WriteLine("### CONNECTING FAILED ###" + Environment.NewLine + exception); + } + + Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###"); + + var last = DateTime.Now; + var msgs = 0; + + while (true) + { + for (int i = 0; i < msgChunkSize; i++) + { + var applicationMessage = new MqttApplicationMessage( + "A/B/C", + Encoding.UTF8.GetBytes("Hello World"), + MqttQualityOfServiceLevel.AtLeastOnce, + false + ); + + //do not await to send as much messages as possible + await client.PublishAsync(applicationMessage); + msgs++; + } + + var now = DateTime.Now; + if (last < now - TimeSpan.FromSeconds(1)) + { + Console.WriteLine( $"sending {msgs} inteded {msgChunkSize / interval.TotalSeconds}" ); + msgs = 0; + last = now; + } + + await Task.Delay(interval).ConfigureAwait(false); + } + } + catch (Exception exception) + { + Console.WriteLine(exception); + } + } + + private static void RunServerAsync() + { + try + { + var options = new MqttServerOptions + { + ConnectionValidator = p => + { + if (p.ClientId == "SpecialClient") + { + if (p.Username != "USER" || p.Password != "PASS") + { + return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; + } + } + + return MqttConnectReturnCode.ConnectionAccepted; + }, + DefaultCommunicationTimeout = TimeSpan.FromMinutes(10) + }; + + var mqttServer = new MqttServerFactory().CreateMqttServer(options); + var last = DateTime.Now; + var msgs = 0; + mqttServer.ApplicationMessageReceived += (sender, args) => + { + msgs++; + var now = DateTime.Now; + if (last < now - TimeSpan.FromSeconds(1)) + { + Console.WriteLine($"received {msgs}"); + msgs = 0; + last = now; + } + }; + mqttServer.Start(); + + Console.WriteLine("Press any key to exit."); + Console.ReadLine(); + + mqttServer.Stop(); + } + catch (Exception e) + { + Console.WriteLine(e); + } + + Console.ReadLine(); + } + } +} diff --git a/Tests/MQTTnet.TestApp.NetFramework/Program.cs b/Tests/MQTTnet.TestApp.NetFramework/Program.cs index 427274f4..d27f423a 100644 --- a/Tests/MQTTnet.TestApp.NetFramework/Program.cs +++ b/Tests/MQTTnet.TestApp.NetFramework/Program.cs @@ -19,6 +19,7 @@ namespace MQTTnet.TestApp.NetFramework Console.WriteLine("MQTTnet - TestApp.NetFramework"); Console.WriteLine("1 = Start client"); Console.WriteLine("2 = Start server"); + Console.WriteLine("3 = Start performance test"); var pressedKey = Console.ReadKey(true); if (pressedKey.Key == ConsoleKey.D1) { @@ -30,6 +31,11 @@ namespace MQTTnet.TestApp.NetFramework Task.Run(() => RunServerAsync(args)); Thread.Sleep(Timeout.Infinite); } + else if (pressedKey.Key == ConsoleKey.D3) + { + Task.Run(() => PerformanceTest.RunAsync()); + Thread.Sleep(Timeout.Infinite); + } } private static async Task RunClientAsync(string[] arguments)