diff --git a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs
index a83fcefd..e8f570c2 100644
--- a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs
+++ b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs
@@ -21,7 +21,10 @@ namespace MQTTnet.AspNetCore.Client
{
var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
var tcpConnection = new TcpConnection(endpoint);
- return new MqttConnectionContext(new MqttPacketFormatterAdapter(options.ProtocolVersion), tcpConnection);
+
+ var writer = new SpanBasedMqttPacketWriter();
+ var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, writer);
+ return new MqttConnectionContext(formatter, tcpConnection);
}
default:
{
diff --git a/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs
index 55192d6c..f5bb0b4c 100644
--- a/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs
+++ b/Source/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs
@@ -5,6 +5,10 @@ using System.Diagnostics;
using System.IO.Pipelines;
using System.Net.Sockets;
+#if NETCOREAPP2_1
+using System.Runtime.InteropServices;
+#endif
+
namespace MQTTnet.AspNetCore.Client.Tcp
{
public class SocketSender
diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
index dc477f76..a1abcc83 100644
--- a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
+++ b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
@@ -1,12 +1,13 @@
- netstandard2.0
+ netstandard2.0;netcoreapp2.1
7.2
+ true
diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
index a0ad1fe0..6d92bf54 100644
--- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
+++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
@@ -17,8 +17,21 @@ namespace MQTTnet.AspNetCore
{
PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
Connection = connection ?? throw new ArgumentNullException(nameof(connection));
+
+ if (Connection.Transport != null)
+ {
+ _input = Connection.Transport.Input;
+ _output = Connection.Transport.Output;
+ }
+
+
+ _reader = new SpanBasedMqttPacketBodyReader();
}
+ private PipeReader _input;
+ private PipeWriter _output;
+ private readonly SpanBasedMqttPacketBodyReader _reader;
+
public string Endpoint => Connection.ConnectionId;
public bool IsSecureConnection => false; // TODO: Fix detection (WS vs. WSS).
@@ -33,20 +46,21 @@ namespace MQTTnet.AspNetCore
private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
- public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
+ public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
if (Connection is TcpConnection tcp && !tcp.IsConnected)
{
- return tcp.StartAsync();
+ await tcp.StartAsync().ConfigureAwait(false);
}
- return Task.CompletedTask;
+ _input = Connection.Transport.Input;
+ _output = Connection.Transport.Output;
}
public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
- Connection.Transport.Input.Complete();
- Connection.Transport.Output.Complete();
+ _input?.Complete();
+ _output?.Complete();
return Task.CompletedTask;
}
@@ -79,7 +93,7 @@ namespace MQTTnet.AspNetCore
{
if (!buffer.IsEmpty)
{
- if (PacketFormatterAdapter.TryDecode(buffer, out var packet, out consumed, out observed))
+ if (PacketFormatterAdapter.TryDecode(_reader, buffer, out var packet, out consumed, out observed))
{
return packet;
}
@@ -114,13 +128,19 @@ namespace MQTTnet.AspNetCore
public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken)
{
- var buffer = PacketFormatterAdapter.Encode(packet).AsMemory();
- var output = Connection.Transport.Output;
+ var formatter = PacketFormatterAdapter;
+
await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
- await output.WriteAsync(buffer, cancellationToken).ConfigureAwait(false);
+ var buffer = formatter.Encode(packet);
+ var msg = buffer.AsMemory();
+ var output = _output;
+ msg.CopyTo(output.GetMemory(msg.Length));
+ PacketFormatterAdapter.FreeBuffer();
+ output.Advance(msg.Length);
+ await output.FlushAsync().ConfigureAwait(false);
}
finally
{
diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs
index ec2528f0..f3cb91a1 100644
--- a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs
+++ b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs
@@ -21,7 +21,9 @@ namespace MQTTnet.AspNetCore
transferFormatFeature.ActiveFormat = TransferFormat.Binary;
}
- using (var adapter = new MqttConnectionContext(new MqttPacketFormatterAdapter(), connection))
+ var writer = new SpanBasedMqttPacketWriter();
+ var formatter = new MqttPacketFormatterAdapter(writer);
+ using (var adapter = new MqttConnectionContext(formatter, connection))
{
var args = new MqttServerAdapterClientAcceptedEventArgs(adapter);
ClientAcceptedHandler?.Invoke(args);
diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
index 3e5e6075..8cb0d6fc 100644
--- a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
+++ b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
@@ -43,8 +43,10 @@ namespace MQTTnet.AspNetCore
var isSecureConnection = clientCertificate != null;
clientCertificate?.Dispose();
+ var writer = new SpanBasedMqttPacketWriter();
+ var formatter = new MqttPacketFormatterAdapter(writer);
var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection);
- var channelAdapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(), _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter)));
+ var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter)));
var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(channelAdapter);
ClientAcceptedHandler?.Invoke(eventArgs);
diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs
index 81d24c0e..8af7ecbe 100644
--- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs
+++ b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs
@@ -9,7 +9,7 @@ namespace MQTTnet.AspNetCore
{
public static class ReaderExtensions
{
- public static bool TryDecode(this MqttPacketFormatterAdapter formatter, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed)
+ public static bool TryDecode(this MqttPacketFormatterAdapter formatter, SpanBasedMqttPacketBodyReader reader, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed)
{
if (formatter == null) throw new ArgumentNullException(nameof(formatter));
@@ -35,18 +35,27 @@ namespace MQTTnet.AspNetCore
}
var bodySlice = copy.Slice(0, bodyLength);
- var buffer = bodySlice.GetArray();
- packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length), buffer.Length + 2));
+ var buffer = bodySlice.GetMemory();
+ reader.SetBuffer(buffer);
+
+ var receivedMqttPacket = new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2);
+
+ if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown)
+ {
+ formatter.DetectProtocolVersion(receivedMqttPacket);
+ }
+
+ packet = formatter.Decode(receivedMqttPacket);
consumed = bodySlice.End;
observed = bodySlice.End;
return true;
}
- private static byte[] GetArray(this in ReadOnlySequence input)
+ private static ReadOnlyMemory GetMemory(this in ReadOnlySequence input)
{
if (input.IsSingleSegment)
{
- return input.First.Span.ToArray();
+ return input.First;
}
// Should be rare
@@ -62,7 +71,7 @@ namespace MQTTnet.AspNetCore
var index = 1;
result = 0;
- var temp = input.Slice(0, Math.Min(5, input.Length)).GetArray();
+ var temp = input.Slice(0, Math.Min(5, input.Length)).GetMemory();
do
{
@@ -70,13 +79,13 @@ namespace MQTTnet.AspNetCore
{
return false;
}
- encodedByte = temp[index];
+ encodedByte = temp.Span[index];
index++;
value += (byte)(encodedByte & 127) * multiplier;
if (multiplier > 128 * 128 * 128)
{
- throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.AsSpan(1, index).ToArray())}).");
+ throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.Slice(1, index).ToArray())}).");
}
multiplier *= 128;
diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs
new file mode 100644
index 00000000..2b118f7b
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs
@@ -0,0 +1,124 @@
+using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
+using System;
+using System.Buffers.Binary;
+using System.Text;
+
+namespace MQTTnet.AspNetCore
+{
+ public class SpanBasedMqttPacketBodyReader : IMqttPacketBodyReader
+ {
+ private ReadOnlyMemory _buffer;
+ private int _offset;
+
+ public void SetBuffer(ReadOnlyMemory buffer)
+ {
+ _buffer = buffer;
+ _offset = 0;
+ }
+
+ public ulong Length => (ulong)_buffer.Length;
+
+ public bool EndOfStream => _buffer.Length.Equals(_offset);
+
+ public ulong Offset => (ulong)_offset;
+
+ public byte ReadByte()
+ {
+ return _buffer.Span[_offset++];
+ }
+
+ public byte[] ReadRemainingData()
+ {
+ return _buffer.Slice(_offset).ToArray();
+ }
+
+ public byte[] ReadWithLengthPrefix()
+ {
+ return ReadSegmentWithLengthPrefix().ToArray();
+ }
+
+ private ReadOnlySpan ReadSegmentWithLengthPrefix()
+ {
+ var span = _buffer.Span;
+ var length = BinaryPrimitives.ReadUInt16BigEndian(span.Slice(_offset));
+
+ var result = span.Slice(_offset+2, length);
+ _offset += 2 + length;
+ return result;
+ }
+
+ public unsafe string ReadStringWithLengthPrefix()
+ {
+ var buffer = ReadSegmentWithLengthPrefix();
+ if (buffer.Length == 0)
+ {
+ return string.Empty;
+ }
+
+ fixed (byte* bytes = &buffer.GetPinnableReference())
+ {
+ var result = Encoding.UTF8.GetString(bytes, buffer.Length);
+ return result;
+ }
+ }
+
+ public ushort ReadTwoByteInteger()
+ {
+ var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset));
+ _offset += 2;
+ return result;
+ }
+
+ public uint ReadFourByteInteger()
+ {
+ var result = BinaryPrimitives.ReadUInt32BigEndian(_buffer.Span.Slice(_offset));
+ _offset += 4;
+ return result;
+ }
+
+ public uint ReadVariableLengthInteger()
+ {
+ var multiplier = 1;
+ var value = 0U;
+ byte encodedByte;
+
+ do
+ {
+ 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;
+ }
+
+ public bool ReadBoolean()
+ {
+ var buffer = ReadByte();
+
+ if (buffer == 0)
+ {
+ return false;
+ }
+
+ if (buffer == 1)
+ {
+ return true;
+ }
+
+ throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only.");
+ }
+
+ public void Seek(ulong position)
+ {
+ _offset = (int)position;
+ }
+ }
+}
diff --git a/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs
new file mode 100644
index 00000000..27e13c49
--- /dev/null
+++ b/Source/MQTTnet.AspnetCore/SpanBasedMqttPacketWriter.cs
@@ -0,0 +1,146 @@
+using MQTTnet.Formatter;
+using System;
+using System.Buffers;
+using System.Buffers.Binary;
+using System.Text;
+
+namespace MQTTnet.AspNetCore
+{
+ public class SpanBasedMqttPacketWriter : IMqttPacketWriter
+ {
+ private readonly ArrayPool _pool;
+
+ public SpanBasedMqttPacketWriter()
+ {
+ _pool = ArrayPool.Create();
+
+ }
+
+ private byte[] _buffer;
+ private int _position;
+
+ public int Length { get; set; }
+
+ public void FreeBuffer()
+ {
+ _pool.Return(_buffer);
+ }
+
+ public byte[] GetBuffer()
+ {
+ return _buffer;
+ }
+
+ public void Reset(int v)
+ {
+ _buffer = _pool.Rent(1500);
+ Length = v;
+ _position = v;
+ }
+
+ public void Seek(int v)
+ {
+ _position = v;
+ }
+
+ public void Write(byte value)
+ {
+ GrowIfNeeded(1);
+ _buffer[_position] = value;
+ Commit(1);
+ }
+
+ public void Write(ushort value)
+ {
+ GrowIfNeeded(2);
+
+ BinaryPrimitives.WriteUInt16BigEndian(_buffer.AsSpan(_position), value);
+ Commit(2);
+ }
+
+ public void Write(IMqttPacketWriter propertyWriter)
+ {
+ if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));
+
+ GrowIfNeeded(propertyWriter.Length);
+ Write(propertyWriter.GetBuffer(), 0, propertyWriter.Length);
+ Commit(propertyWriter.Length);
+ }
+
+ public void Write(byte[] payload, int start, int length)
+ {
+ GrowIfNeeded(length);
+
+ payload.AsSpan(start, length).CopyTo(_buffer.AsSpan(_position));
+ Commit(length);
+ }
+
+ public void WriteVariableLengthInteger(uint value)
+ {
+ GrowIfNeeded(4);
+
+ var x = value;
+ do
+ {
+ var encodedByte = x % 128;
+ x = x / 128;
+ if (x > 0)
+ {
+ encodedByte = encodedByte | 128;
+ }
+
+ _buffer[_position] = (byte)encodedByte;
+ Commit(1);
+ } while (x > 0);
+ }
+
+ public void WriteWithLengthPrefix(string value)
+ {
+ var bytesLength = Encoding.UTF8.GetByteCount(value ?? string.Empty);
+ GrowIfNeeded(bytesLength + 2);
+
+ Write((ushort)bytesLength);
+ Encoding.UTF8.GetBytes(value ?? string.Empty, 0, value?.Length ?? 0, _buffer, _position);
+ Commit(bytesLength);
+ }
+
+ public void WriteWithLengthPrefix(byte[] payload)
+ {
+ GrowIfNeeded(payload.Length + 2);
+
+ Write((ushort)payload.Length);
+ payload.CopyTo(_buffer, _position);
+ Commit(payload.Length);
+ }
+
+ private void Commit(int count)
+ {
+ if (_position == Length)
+ {
+ Length += count;
+ }
+
+ _position += count;
+ }
+
+ private void GrowIfNeeded(int requiredAdditional)
+ {
+ var requiredTotal = _position + requiredAdditional;
+ if (_buffer.Length >= requiredTotal)
+ {
+ return;
+ }
+
+ var newBufferLength = _buffer.Length;
+ while (newBufferLength < requiredTotal)
+ {
+ newBufferLength *= 2;
+ }
+
+ var newBuffer = _pool.Rent(newBufferLength);
+ Array.Copy(_buffer, newBuffer, _buffer.Length);
+ _pool.Return(_buffer);
+ _buffer = newBuffer;
+ }
+ }
+}
diff --git a/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs b/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs
index 683d6b64..355ad96f 100644
--- a/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs
+++ b/Source/MQTTnet/Adapter/ReceivedMqttPacket.cs
@@ -4,16 +4,16 @@ namespace MQTTnet.Adapter
{
public class ReceivedMqttPacket
{
- public ReceivedMqttPacket(byte fixedHeader, MqttPacketBodyReader body, int totalLength)
+ public ReceivedMqttPacket(byte fixedHeader, IMqttPacketBodyReader body, int totalLength)
{
FixedHeader = fixedHeader;
Body = body;
TotalLength = totalLength;
}
- public byte FixedHeader { get; }
+ public byte FixedHeader { get; set; }
- public MqttPacketBodyReader Body { get; }
+ public IMqttPacketBodyReader Body { get; }
public int TotalLength { get; }
}
diff --git a/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs
new file mode 100644
index 00000000..632c4931
--- /dev/null
+++ b/Source/MQTTnet/Formatter/IMqttPacketBodyReader.cs
@@ -0,0 +1,31 @@
+using System;
+
+namespace MQTTnet.Formatter
+{
+ public interface IMqttPacketBodyReader
+ {
+ ulong Length { get; }
+
+ ulong Offset { get; }
+
+ bool EndOfStream { get; }
+
+ byte ReadByte();
+
+ byte[] ReadRemainingData();
+
+ ushort ReadTwoByteInteger();
+
+ string ReadStringWithLengthPrefix();
+
+ byte[] ReadWithLengthPrefix();
+
+ uint ReadFourByteInteger();
+
+ uint ReadVariableLengthInteger();
+
+ bool ReadBoolean();
+
+ void Seek(ulong position);
+ }
+}
diff --git a/Source/MQTTnet/Formatter/IMqttPacketWriter.cs b/Source/MQTTnet/Formatter/IMqttPacketWriter.cs
new file mode 100644
index 00000000..cce98cf5
--- /dev/null
+++ b/Source/MQTTnet/Formatter/IMqttPacketWriter.cs
@@ -0,0 +1,20 @@
+namespace MQTTnet.Formatter
+{
+ public interface IMqttPacketWriter
+ {
+ int Length { get; }
+
+ void WriteWithLengthPrefix(string value);
+ void Write(byte returnCode);
+ void WriteWithLengthPrefix(byte[] payload);
+ void Write(ushort keepAlivePeriod);
+
+ void Write(IMqttPacketWriter propertyWriter);
+ void WriteVariableLengthInteger(uint length);
+ void Write(byte[] payload, int v, int length);
+ void Reset(int v);
+ void Seek(int v);
+ void FreeBuffer();
+ byte[] GetBuffer();
+ }
+}
diff --git a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs
index db971f09..c45fb3d9 100644
--- a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs
+++ b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs
@@ -1,11 +1,12 @@
using System;
+using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using MQTTnet.Exceptions;
namespace MQTTnet.Formatter
{
- public class MqttPacketBodyReader
+ public class MqttPacketBodyReader : IMqttPacketBodyReader
{
private readonly byte[] _buffer;
private readonly ulong _initialOffset;
@@ -68,9 +69,9 @@ namespace MQTTnet.Formatter
throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only.");
}
- public ArraySegment ReadRemainingData()
+ public byte[] ReadRemainingData()
{
- return new ArraySegment(_buffer, (int)Offset, (int)(_length - Offset));
+ return new ArraySegment(_buffer, (int)Offset, (int)(_length - Offset)).ToArray();
}
public ushort ReadTwoByteInteger()
@@ -95,7 +96,12 @@ namespace MQTTnet.Formatter
return (uint)(byte0 << 24 | byte1 << 16 | byte2 << 8 | byte3);
}
- public ArraySegment ReadWithLengthPrefix()
+ public byte[] ReadWithLengthPrefix()
+ {
+ return ReadSegmentWithLengthPrefix().ToArray();
+ }
+
+ private ArraySegment ReadSegmentWithLengthPrefix()
{
var length = ReadTwoByteInteger();
@@ -140,7 +146,7 @@ namespace MQTTnet.Formatter
public string ReadStringWithLengthPrefix()
{
- var buffer = ReadWithLengthPrefix();
+ var buffer = ReadSegmentWithLengthPrefix();
return Encoding.UTF8.GetString(buffer.Array, buffer.Offset, buffer.Count);
}
}
diff --git a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs
index 47f80db9..723b6b8c 100644
--- a/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs
+++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs
@@ -12,14 +12,26 @@ namespace MQTTnet.Formatter
private IMqttPacketFormatter _formatter;
public MqttPacketFormatterAdapter()
+ : this(new MqttPacketWriter())
+ {
+ }
+
+ public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion)
+ : this(protocolVersion, new MqttPacketWriter())
{
}
- public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion)
+ public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion, IMqttPacketWriter writer)
+ : this(writer)
{
UseProtocolVersion(protocolVersion);
}
+ public MqttPacketFormatterAdapter(IMqttPacketWriter writer)
+ {
+ Writer = writer;
+ }
+
public MqttProtocolVersion ProtocolVersion { get; private set; } = MqttProtocolVersion.Unknown;
public IMqttDataConverter DataConverter
@@ -31,6 +43,8 @@ namespace MQTTnet.Formatter
return _formatter.DataConverter;
}
}
+
+ public IMqttPacketWriter Writer { get; }
public ArraySegment Encode(MqttBasePacket packet)
{
@@ -75,28 +89,30 @@ namespace MQTTnet.Formatter
}
ProtocolVersion = protocolVersion;
+ _formatter = GetMqttPacketFormatter(protocolVersion, Writer);
+ }
+ public static IMqttPacketFormatter GetMqttPacketFormatter(MqttProtocolVersion protocolVersion, IMqttPacketWriter writer)
+ {
+ if (protocolVersion == MqttProtocolVersion.Unknown)
+ {
+ throw new InvalidOperationException("MQTT protocol version is invalid.");
+ }
+
switch (protocolVersion)
{
case MqttProtocolVersion.V500:
{
- _formatter = new MqttV500PacketFormatter();
- break;
+ return new MqttV500PacketFormatter(writer);
}
-
case MqttProtocolVersion.V311:
{
- _formatter = new MqttV311PacketFormatter();
- break;
+ return new MqttV311PacketFormatter(writer);
}
-
case MqttProtocolVersion.V310:
{
-
- _formatter = new MqttV310PacketFormatter();
- break;
+ return new MqttV310PacketFormatter(writer);
}
-
default:
{
throw new NotSupportedException();
diff --git a/Source/MQTTnet/Formatter/MqttPacketWriter.cs b/Source/MQTTnet/Formatter/MqttPacketWriter.cs
index 1cedd080..eb7253e0 100644
--- a/Source/MQTTnet/Formatter/MqttPacketWriter.cs
+++ b/Source/MQTTnet/Formatter/MqttPacketWriter.cs
@@ -11,7 +11,7 @@ namespace MQTTnet.Formatter
/// same as for the original MemoryStream in .net. Also this implementation allows accessing the internal
/// buffer for all platforms and .net framework versions (which is not available at the regular MemoryStream).
///
- public class MqttPacketWriter
+ public class MqttPacketWriter : IMqttPacketWriter
{
private static readonly ArraySegment ZeroVariableLengthIntegerArray = new ArraySegment(new byte[1], 0, 1);
private static readonly ArraySegment ZeroTwoByteIntegerArray = new ArraySegment(new byte[2], 0, 2);
@@ -33,6 +33,19 @@ namespace MQTTnet.Formatter
return (byte)fixedHeader;
}
+ public static int GetLengthOfVariableInteger(uint value)
+ {
+ var result = 0;
+ var x = value;
+ do
+ {
+ x = x / 128;
+ result++;
+ } while (x > 0);
+
+ return result;
+ }
+
public static ArraySegment EncodeVariableLengthInteger(uint value)
{
if (value == 0)
@@ -129,16 +142,22 @@ namespace MQTTnet.Formatter
IncreasePosition(count);
}
- public void Write(MqttPacketWriter propertyWriter)
+ public void Write(IMqttPacketWriter propertyWriter)
{
if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));
- if (propertyWriter.Length == 0)
+ if (propertyWriter is MqttPacketWriter writer)
{
+ if (writer.Length == 0)
+ {
+ return;
+ }
+
+ Write(writer._buffer, 0, writer.Length);
return;
}
- Write(propertyWriter._buffer, 0, propertyWriter.Length);
+ throw new InvalidOperationException($"{nameof(propertyWriter)} must be of type {typeof(MqttPacketWriter).Name}");
}
public void Reset(int length)
diff --git a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs
index 02a28a12..70af10e6 100644
--- a/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs
+++ b/Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs
@@ -2,7 +2,6 @@
using System.Linq;
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
-using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -11,8 +10,20 @@ namespace MQTTnet.Formatter.V3
public class MqttV310PacketFormatter : IMqttPacketFormatter
{
private const int FixedHeaderSize = 1;
+
+ private readonly IMqttPacketWriter _packetWriter;
+
+ public MqttV310PacketFormatter()
+ : this(new MqttPacketWriter())
+ {
+
+ }
+
+ public MqttV310PacketFormatter(IMqttPacketWriter packetWriter)
+ {
+ _packetWriter = packetWriter;
+ }
- private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
public IMqttDataConverter DataConverter { get; } = new MqttV310DataConverter();
@@ -27,17 +38,18 @@ namespace MQTTnet.Formatter.V3
var fixedHeader = EncodePacket(packet, _packetWriter);
var remainingLength = (uint)(_packetWriter.Length - 5);
- var remainingLengthBuffer = MqttPacketWriter.EncodeVariableLengthInteger(remainingLength);
+ var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
- var headerSize = FixedHeaderSize + remainingLengthBuffer.Count;
+ var headerSize = FixedHeaderSize + remainingLengthSize;
var headerOffset = 5 - headerSize;
// Position cursor on correct offset on beginning of array (has leading 0x0)
_packetWriter.Seek(headerOffset);
_packetWriter.Write(fixedHeader);
- _packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count);
+ _packetWriter.WriteVariableLengthInteger(remainingLength);
var buffer = _packetWriter.GetBuffer();
+
return new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset);
}
@@ -77,7 +89,7 @@ namespace MQTTnet.Formatter.V3
_packetWriter.FreeBuffer();
}
- private byte EncodePacket(MqttBasePacket packet, MqttPacketWriter packetWriter)
+ private byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter)
{
switch (packet)
{
@@ -100,7 +112,7 @@ namespace MQTTnet.Formatter.V3
}
}
- private static MqttBasePacket DecodeUnsubAckPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -110,7 +122,7 @@ namespace MQTTnet.Formatter.V3
};
}
- private static MqttBasePacket DecodePubCompPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -120,7 +132,7 @@ namespace MQTTnet.Formatter.V3
};
}
- private static MqttBasePacket DecodePubRelPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -130,7 +142,7 @@ namespace MQTTnet.Formatter.V3
};
}
- private static MqttBasePacket DecodePubRecPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -140,7 +152,7 @@ namespace MQTTnet.Formatter.V3
};
}
- private static MqttBasePacket DecodePubAckPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -150,7 +162,7 @@ namespace MQTTnet.Formatter.V3
};
}
- private static MqttBasePacket DecodeUnsubscribePacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -167,7 +179,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
- private static MqttBasePacket DecodeSubscribePacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -223,7 +235,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
- private MqttBasePacket DecodeConnectPacket(MqttPacketBodyReader body)
+ private MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -264,7 +276,7 @@ namespace MQTTnet.Formatter.V3
packet.WillMessage = new MqttApplicationMessage
{
Topic = body.ReadStringWithLengthPrefix(),
- Payload = body.ReadWithLengthPrefix().ToArray(),
+ Payload = body.ReadWithLengthPrefix(),
QualityOfServiceLevel = (MqttQualityOfServiceLevel)willQoS,
Retain = willRetain
};
@@ -284,7 +296,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
- private static MqttBasePacket DecodeSubAckPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -301,7 +313,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
- protected virtual MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
+ protected virtual MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -332,7 +344,7 @@ namespace MQTTnet.Formatter.V3
}
}
- protected virtual byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
+ protected virtual byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter)
{
ValidateConnectPacket(packet);
@@ -394,7 +406,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
}
- protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
+ protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter)
{
packetWriter.Write(0); // Reserved.
packetWriter.Write((byte)packet.ReturnCode.Value);
@@ -402,7 +414,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
- private static byte EncodePubRelPacket(MqttPubRelPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -414,7 +426,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
}
- private static byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter)
{
ValidatePublishPacket(packet);
@@ -459,7 +471,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
}
- private static byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -471,7 +483,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
}
- private static byte EncodePubRecPacket(MqttPubRecPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -483,7 +495,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
}
- private static byte EncodePubCompPacket(MqttPubCompPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -495,7 +507,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
}
- private static byte EncodeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
@@ -518,7 +530,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02);
}
- private static byte EncodeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -538,7 +550,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck);
}
- private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
@@ -560,7 +572,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
}
- private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -577,7 +589,7 @@ namespace MQTTnet.Formatter.V3
}
// ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
- protected static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body)
+ protected static void ThrowIfBodyIsEmpty(IMqttPacketBodyReader body)
{
if (body == null || body.Length == 0)
{
diff --git a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs
index 0bc60a99..a846cd09 100644
--- a/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs
+++ b/Source/MQTTnet/Formatter/V3/MqttV311PacketFormatter.cs
@@ -6,7 +6,17 @@ namespace MQTTnet.Formatter.V3
{
public class MqttV311PacketFormatter : MqttV310PacketFormatter
{
- protected override byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
+ public MqttV311PacketFormatter()
+ : base()
+ {
+ }
+
+ public MqttV311PacketFormatter(IMqttPacketWriter packetWriter)
+ : base(packetWriter)
+ {
+ }
+
+ protected override byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter)
{
ValidateConnectPacket(packet);
@@ -68,7 +78,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
}
- protected override byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
+ protected override byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter)
{
byte connectAcknowledgeFlags = 0x0;
if (packet.IsSessionPresent)
@@ -82,7 +92,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
- protected override MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
+ protected override MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs
index 54588eea..46692cca 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs
@@ -41,7 +41,7 @@ namespace MQTTnet.Formatter.V5
}
}
- private static MqttBasePacket DecodeConnectPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -134,7 +134,7 @@ namespace MQTTnet.Formatter.V5
if (packet.WillMessage != null)
{
packet.WillMessage.Topic = body.ReadStringWithLengthPrefix();
- packet.WillMessage.Payload = body.ReadWithLengthPrefix().ToArray();
+ packet.WillMessage.Payload = body.ReadWithLengthPrefix();
}
if (usernameFlag)
@@ -150,7 +150,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -239,7 +239,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodeDisconnectPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeDisconnectPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -277,7 +277,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodeSubscribePacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -327,7 +327,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodeSubAckPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -363,7 +363,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodeUnsubscribePacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -394,7 +394,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodeUnsubAckPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -440,7 +440,7 @@ namespace MQTTnet.Formatter.V5
return new MqttPingRespPacket();
}
- private static MqttBasePacket DecodePublishPacket(byte header, MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePublishPacket(byte header, IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -511,7 +511,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodePubAckPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -549,7 +549,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodePubRecPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -587,7 +587,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodePubRelPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -625,7 +625,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodePubCompPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -663,7 +663,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
- private static MqttBasePacket DecodeAuthPacket(MqttPacketBodyReader body)
+ private static MqttBasePacket DecodeAuthPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -709,7 +709,7 @@ namespace MQTTnet.Formatter.V5
}
// ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
- private static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body)
+ private static void ThrowIfBodyIsEmpty(IMqttPacketBodyReader body)
{
if (body == null || body.Length == 0)
{
diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs
index 83f003be..5ccd0301 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs
@@ -8,7 +8,19 @@ namespace MQTTnet.Formatter.V5
{
public class MqttV500PacketEncoder
{
- private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
+ private readonly IMqttPacketWriter _packetWriter;
+
+ public MqttV500PacketEncoder()
+ : this(new MqttPacketWriter())
+ {
+
+ }
+
+ public MqttV500PacketEncoder(IMqttPacketWriter packetWriter)
+ {
+ _packetWriter = packetWriter;
+ }
+
public ArraySegment Encode(MqttBasePacket packet)
{
@@ -21,15 +33,15 @@ namespace MQTTnet.Formatter.V5
var fixedHeader = EncodePacket(packet, _packetWriter);
var remainingLength = (uint)(_packetWriter.Length - 5);
- var remainingLengthBuffer = MqttPacketWriter.EncodeVariableLengthInteger(remainingLength);
-
- var headerSize = 1 + remainingLengthBuffer.Count;
+ var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
+
+ var headerSize = 1 + remainingLengthSize;
var headerOffset = 5 - headerSize;
// Position cursor on correct offset on beginning of array (has leading 0x0)
_packetWriter.Seek(headerOffset);
_packetWriter.Write(fixedHeader);
- _packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count);
+ _packetWriter.WriteVariableLengthInteger(remainingLength);
var buffer = _packetWriter.GetBuffer();
return new ArraySegment(buffer, headerOffset, _packetWriter.Length - headerOffset);
@@ -40,7 +52,7 @@ namespace MQTTnet.Formatter.V5
_packetWriter.FreeBuffer();
}
- private static byte EncodePacket(MqttBasePacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter)
{
switch (packet)
{
@@ -64,7 +76,7 @@ namespace MQTTnet.Formatter.V5
}
}
- private static byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -150,7 +162,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
}
- private static byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -195,7 +207,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
- private static byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -261,7 +273,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
}
- private static byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -294,7 +306,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
}
- private static byte EncodePubRecPacket(MqttPubRecPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter)
{
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -322,7 +334,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
}
- private static byte EncodePubRelPacket(MqttPubRelPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter)
{
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -350,7 +362,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
}
- private static byte EncodePubCompPacket(MqttPubCompPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter)
{
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -378,7 +390,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
}
- private static byte EncodeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
@@ -425,7 +437,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02);
}
- private static byte EncodeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
@@ -450,7 +462,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck);
}
- private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
@@ -474,7 +486,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
}
- private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
@@ -499,7 +511,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck);
}
- private static byte EncodeDisconnectPacket(MqttDisconnectPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeDisconnectPacket(MqttDisconnectPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.ReasonCode.HasValue)
{
@@ -532,7 +544,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PingResp);
}
- private static byte EncodeAuthPacket(MqttAuthPacket packet, MqttPacketWriter packetWriter)
+ private static byte EncodeAuthPacket(MqttAuthPacket packet, IMqttPacketWriter packetWriter)
{
packetWriter.Write((byte)packet.ReasonCode);
diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs
index ab52885e..9a52a98d 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs
@@ -6,9 +6,19 @@ namespace MQTTnet.Formatter.V5
{
public class MqttV500PacketFormatter : IMqttPacketFormatter
{
- private readonly MqttV500PacketEncoder _encoder = new MqttV500PacketEncoder();
+ private readonly MqttV500PacketEncoder _encoder;
private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder();
+ public MqttV500PacketFormatter()
+ {
+ _encoder = new MqttV500PacketEncoder();
+ }
+
+ public MqttV500PacketFormatter(IMqttPacketWriter writer)
+ {
+ _encoder = new MqttV500PacketEncoder(writer);
+ }
+
public IMqttDataConverter DataConverter { get; } = new MqttV500DataConverter();
public ArraySegment Encode(MqttBasePacket mqttPacket)
diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs
index c7f28dd4..e30cb3f2 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs
@@ -9,11 +9,11 @@ namespace MQTTnet.Formatter.V5
{
public class MqttV500PropertiesReader
{
- private readonly MqttPacketBodyReader _body;
+ private readonly IMqttPacketBodyReader _body;
private readonly uint _length;
private readonly ulong _targetOffset;
- public MqttV500PropertiesReader(MqttPacketBodyReader body)
+ public MqttV500PropertiesReader(IMqttPacketBodyReader body)
{
_body = body ?? throw new ArgumentNullException(nameof(body));
@@ -75,7 +75,7 @@ namespace MQTTnet.Formatter.V5
public byte[] ReadAuthenticationData()
{
- return _body.ReadWithLengthPrefix().ToArray();
+ return _body.ReadWithLengthPrefix();
}
public bool? ReadRetainAvailable()
@@ -165,7 +165,7 @@ namespace MQTTnet.Formatter.V5
public byte[] ReadCorrelationData()
{
- return _body.ReadWithLengthPrefix().ToArray();
+ return _body.ReadWithLengthPrefix();
}
public string ReadContentType()
diff --git a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs
index e10790ce..9e5fe39a 100644
--- a/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs
+++ b/Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs
@@ -63,7 +63,7 @@ namespace MQTTnet.Formatter.V5
Write(MqttPropertyId.AuthenticationMethod, value);
}
- public void WriteToPacket(MqttPacketWriter packetWriter)
+ public void WriteToPacket(IMqttPacketWriter packetWriter)
{
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
diff --git a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj
index e523de5d..06f0e5f1 100644
--- a/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj
+++ b/Tests/MQTTnet.AspNetCore.Tests/MQTTnet.AspNetCore.Tests.csproj
@@ -14,7 +14,7 @@
-
+
diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs
new file mode 100644
index 00000000..2d389ab1
--- /dev/null
+++ b/Tests/MQTTnet.AspNetCore.Tests/MqttPacketSerializerTests.cs
@@ -0,0 +1,22 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Formatter;
+using MQTTnet.Tests;
+
+namespace MQTTnet.AspNetCore.Tests
+{
+ [TestClass]
+ public class MqttPacketSerializerTestsWithSpanBasedReader : MqttPacketSerializer_Tests
+ {
+ protected override IMqttPacketBodyReader ReaderFactory(byte[] data)
+ {
+ var result = new SpanBasedMqttPacketBodyReader();
+ result.SetBuffer(data);
+ return result;
+ }
+
+ protected override IMqttPacketWriter WriterFactory()
+ {
+ return new SpanBasedMqttPacketWriter();
+ }
+ }
+}
diff --git a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs
index dedeff33..4d69d37c 100644
--- a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs
+++ b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs
@@ -24,21 +24,23 @@ namespace MQTTnet.AspNetCore.Tests
var observed = part.Start;
var result = false;
+ var reader = new SpanBasedMqttPacketBodyReader();
+
part = sequence.Slice(sequence.Start, 0); // empty message should fail
- result = serializer.TryDecode(part, out packet, out consumed, out observed);
+ result = serializer.TryDecode(reader, part, out packet, out consumed, out observed);
Assert.IsFalse(result);
part = sequence.Slice(sequence.Start, 1); // partial fixed header should fail
- result = serializer.TryDecode(part, out packet, out consumed, out observed);
+ result = serializer.TryDecode(reader, part, out packet, out consumed, out observed);
Assert.IsFalse(result);
part = sequence.Slice(sequence.Start, 4); // partial body should fail
- result = serializer.TryDecode(part, out packet, out consumed, out observed);
+ result = serializer.TryDecode(reader, part, out packet, out consumed, out observed);
Assert.IsFalse(result);
part = sequence; // complete msg should work
- result = serializer.TryDecode(part, out packet, out consumed, out observed);
+ result = serializer.TryDecode(reader, part, out packet, out consumed, out observed);
Assert.IsTrue(result);
}
}
diff --git a/Tests/MQTTnet.Benchmarks/AllowNonOptimized.cs b/Tests/MQTTnet.Benchmarks/Configurations/AllowNonOptimized.cs
similarity index 54%
rename from Tests/MQTTnet.Benchmarks/AllowNonOptimized.cs
rename to Tests/MQTTnet.Benchmarks/Configurations/AllowNonOptimized.cs
index ecb31a20..076d1e8d 100644
--- a/Tests/MQTTnet.Benchmarks/AllowNonOptimized.cs
+++ b/Tests/MQTTnet.Benchmarks/Configurations/AllowNonOptimized.cs
@@ -3,20 +3,17 @@ using BenchmarkDotNet.Configs;
using BenchmarkDotNet.Jobs;
using BenchmarkDotNet.Validators;
-namespace MQTTnet.Benchmarks
+namespace MQTTnet.Benchmarks.Configurations
{
///
/// this options may be used to run benchmarks in debugmode and attach a performance profiler
/// https://benchmarkdotnet.org/Configs/Configs.htm
///
- public class AllowNonOptimized : ManualConfig
+ public class AllowNonOptimized : BaseConfig
{
public AllowNonOptimized()
{
Add(JitOptimizationsValidator.DontFailOnError); // ALLOW NON-OPTIMIZED DLLS
- Add(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default
- Add(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default
- Add(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default
Add(Job.InProcess);
}
}
diff --git a/Tests/MQTTnet.Benchmarks/Configurations/BaseConfig.cs b/Tests/MQTTnet.Benchmarks/Configurations/BaseConfig.cs
new file mode 100644
index 00000000..418bb2f3
--- /dev/null
+++ b/Tests/MQTTnet.Benchmarks/Configurations/BaseConfig.cs
@@ -0,0 +1,15 @@
+using BenchmarkDotNet.Configs;
+using System.Linq;
+
+namespace MQTTnet.Benchmarks.Configurations
+{
+ public class BaseConfig : ManualConfig
+ {
+ public BaseConfig()
+ {
+ Add(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default
+ Add(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default
+ Add(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default
+ }
+ }
+}
diff --git a/Tests/MQTTnet.Benchmarks/Configurations/RuntimeCompareConfig.cs b/Tests/MQTTnet.Benchmarks/Configurations/RuntimeCompareConfig.cs
new file mode 100644
index 00000000..608271e2
--- /dev/null
+++ b/Tests/MQTTnet.Benchmarks/Configurations/RuntimeCompareConfig.cs
@@ -0,0 +1,17 @@
+using BenchmarkDotNet.Configs;
+using BenchmarkDotNet.Environments;
+using BenchmarkDotNet.Jobs;
+using BenchmarkDotNet.Toolchains.CsProj;
+
+namespace MQTTnet.Benchmarks.Configurations
+{
+ public class RuntimeCompareConfig : BaseConfig
+ {
+ public RuntimeCompareConfig()
+ {
+ Add(Job.Default.With(Runtime.Clr));
+ Add(Job.Default.With(Runtime.Core).With(CsProjCoreToolchain.NetCoreApp21));
+ }
+
+ }
+}
diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj
index 02818bf4..03aabc85 100644
--- a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj
+++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj
@@ -3,7 +3,7 @@
Exe
Full
- net461
+ net461;netcoreapp2.1
netcoreapp2.1
7.2
diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs
index 9407ccfe..280416d3 100644
--- a/Tests/MQTTnet.Benchmarks/Program.cs
+++ b/Tests/MQTTnet.Benchmarks/Program.cs
@@ -1,5 +1,6 @@
using System;
using BenchmarkDotNet.Running;
+using MQTTnet.Benchmarks.Configurations;
using MQTTnet.Diagnostics;
namespace MQTTnet.Benchmarks
@@ -43,7 +44,7 @@ namespace MQTTnet.Benchmarks
BenchmarkRunner.Run();
break;
case '8':
- BenchmarkRunner.Run(new AllowNonOptimized());
+ BenchmarkRunner.Run(new RuntimeCompareConfig()/*new AllowNonOptimized()*/);
break;
}
diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs
index a1aad718..429a2312 100644
--- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs
+++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs
@@ -231,6 +231,32 @@ namespace MQTTnet.Tests
SerializeAndCompare(p, "Ow4ABUEvQi9DAHtIRUxMTw==");
}
+ [TestMethod]
+ public void SerializeV500_MqttPublishPacket()
+ {
+ var prop = new MqttPublishPacketProperties();
+
+ prop.ResponseTopic = "/Response";
+
+ prop.UserProperties.Add(new MqttUserProperty("Foo", "Bar"));
+
+ var p = new MqttPublishPacket
+ {
+ PacketIdentifier = 123,
+ Dup = true,
+ Retain = true,
+ Payload = Encoding.ASCII.GetBytes("HELLO"),
+ QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
+ Topic = "A/B/C",
+ Properties = prop
+ };
+
+ var deserialized = Roundtrip(p, MqttProtocolVersion.V500);
+
+ Assert.AreEqual(prop.ResponseTopic, deserialized.Properties.ResponseTopic);
+ Assert.IsTrue(deserialized.Properties.UserProperties.Any(x => x.Name == "Foo"));
+ }
+
[TestMethod]
public void DeserializeV311_MqttPublishPacket()
{
@@ -528,21 +554,21 @@ namespace MQTTnet.Tests
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data)));
}
- private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
+ protected virtual IMqttPacketWriter WriterFactory()
{
- IMqttPacketFormatter serializer;
- if (protocolVersion == MqttProtocolVersion.V311)
- {
- serializer = new MqttV311PacketFormatter();
- }
- else if (protocolVersion == MqttProtocolVersion.V310)
- {
- serializer = new MqttV310PacketFormatter();
- }
- else
- {
- throw new NotSupportedException();
- }
+ return new MqttPacketWriter();
+ }
+
+ protected virtual IMqttPacketBodyReader ReaderFactory(byte[] data)
+ {
+ return new MqttPacketBodyReader(data, 0, data.Length);
+ }
+
+ private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
+ {
+ var writer = WriterFactory();
+
+ var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
var buffer1 = serializer.Encode(packet);
@@ -554,7 +580,8 @@ namespace MQTTnet.Tests
using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength))
{
- var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length), 0));
+ var reader = ReaderFactory(bodyStream.ToArray());
+ var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0));
var buffer2 = serializer.Encode(deserializedPacket);
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));
@@ -562,23 +589,13 @@ namespace MQTTnet.Tests
}
}
- private static T Roundtrip(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
+ private T Roundtrip(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
where T : MqttBasePacket
{
- IMqttPacketFormatter serializer;
- if (protocolVersion == MqttProtocolVersion.V311)
- {
- serializer = new MqttV311PacketFormatter();
- }
- else if (protocolVersion == MqttProtocolVersion.V310)
- {
- serializer = new MqttV310PacketFormatter();
- }
- else
- {
- throw new NotSupportedException();
- }
+ var writer = WriterFactory();
+ var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
+
var buffer1 = serializer.Encode(packet);
using (var headerStream = new MemoryStream(Join(buffer1)))
@@ -590,7 +607,8 @@ namespace MQTTnet.Tests
using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, (int)header.RemainingLength))
{
- return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length), 0));
+ var reader = ReaderFactory(bodyStream.ToArray());
+ return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0));
}
}
}