diff --git a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs index b5801f46..93314f44 100644 --- a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs +++ b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs @@ -20,7 +20,7 @@ namespace MQTTnet.AspNetCore.Client { var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort()); var tcpConnection = new TcpConnection(endpoint); - return new MqttConnectionContext(new MqttPacketSerializerAdapter(options.ProtocolVersion), tcpConnection); + return new MqttConnectionContext(new MqttPacketFormatterAdapter(options.ProtocolVersion), tcpConnection); } default: { diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index aa2d7d8c..b24f6e1d 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -13,15 +13,15 @@ namespace MQTTnet.AspNetCore { public class MqttConnectionContext : IMqttChannelAdapter { - public MqttConnectionContext(MqttPacketSerializerAdapter packetSerializerAdapter, ConnectionContext connection) + public MqttConnectionContext(MqttPacketFormatterAdapter packetFormatterAdapter, ConnectionContext connection) { - PacketSerializerAdapter = packetSerializerAdapter ?? throw new ArgumentNullException(nameof(packetSerializerAdapter)); + PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter)); Connection = connection ?? throw new ArgumentNullException(nameof(connection)); } public string Endpoint => Connection.ConnectionId; public ConnectionContext Connection { get; } - public MqttPacketSerializerAdapter PacketSerializerAdapter { get; } + public MqttPacketFormatterAdapter PacketFormatterAdapter { get; } public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; @@ -72,7 +72,7 @@ namespace MQTTnet.AspNetCore { if (!buffer.IsEmpty) { - if (PacketSerializerAdapter.Formatter.TryDeserialize(buffer, out var packet, out consumed, out observed)) + if (PacketFormatterAdapter.TryDecode(buffer, out var packet, out consumed, out observed)) { return packet; } @@ -107,7 +107,7 @@ namespace MQTTnet.AspNetCore public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) { - var buffer = PacketSerializerAdapter.Encode(packet).AsMemory(); + var buffer = PacketFormatterAdapter.Encode(packet).AsMemory(); var output = Connection.Transport.Output; await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs index f533ed7c..0c09a7a9 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs @@ -21,7 +21,7 @@ namespace MQTTnet.AspNetCore transferFormatFeature.ActiveFormat = TransferFormat.Binary; } - using (var adapter = new MqttConnectionContext(new MqttPacketSerializerAdapter(), connection)) + using (var adapter = new MqttConnectionContext(new MqttPacketFormatterAdapter(), connection)) { var args = new MqttServerAdapterClientAcceptedEventArgs(adapter); ClientAccepted?.Invoke(this, args); diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index ee310222..65514d0b 100644 --- a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs +++ b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs @@ -27,7 +27,7 @@ namespace MQTTnet.AspNetCore { if (webSocket == null) throw new ArgumentNullException(nameof(webSocket)); - var clientAdapter = new MqttChannelAdapter(new MqttWebSocketChannel(webSocket, endpoint), new MqttPacketSerializerAdapter(), new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); + var clientAdapter = new MqttChannelAdapter(new MqttWebSocketChannel(webSocket, endpoint), new MqttPacketFormatterAdapter(), new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter); ClientAccepted?.Invoke(this, eventArgs); diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index d7c192b9..abe8ef7b 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 TryDeserialize(this IMqttPacketFormatter formatter, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed) + public static bool TryDecode(this MqttPacketFormatterAdapter formatter, in ReadOnlySequence input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed) { if (formatter == null) throw new ArgumentNullException(nameof(formatter)); @@ -17,6 +17,7 @@ namespace MQTTnet.AspNetCore consumed = input.Start; observed = input.End; var copy = input; + if (copy.Length < 2) { return false; diff --git a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs index 82003204..183badb4 100644 --- a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Adapter { string Endpoint { get; } - MqttPacketSerializerAdapter PacketSerializerAdapter { get; } + MqttPacketFormatterAdapter PacketFormatterAdapter { get; } event EventHandler ReadingPacketStarted; diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 463b898a..340a528b 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -16,31 +16,33 @@ namespace MQTTnet.Adapter public class MqttChannelAdapter : IMqttChannelAdapter { private const uint ErrorOperationAborted = 0x800703E3; - private const int ReadBufferSize = 4096; // TODO: Move buffer size to config + private const uint ReadBufferSize = 4096; // TODO: Move buffer size to config private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1); private readonly IMqttNetChildLogger _logger; private readonly IMqttChannel _channel; - - private readonly byte[] _fixedHeaderBuffer = new byte[2]; - private readonly byte[] _singleByteBuffer = new byte[1]; + private readonly MqttPacketReader _packetReader; + private readonly byte[] _fixedHeaderBuffer = new byte[2]; + private bool _isDisposed; - public MqttChannelAdapter(IMqttChannel channel, MqttPacketSerializerAdapter packetSerializerAdapter, IMqttNetChildLogger logger) + public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetChildLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); _channel = channel ?? throw new ArgumentNullException(nameof(channel)); - PacketSerializerAdapter = packetSerializerAdapter ?? throw new ArgumentNullException(nameof(packetSerializerAdapter)); + PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter)); + _packetReader = new MqttPacketReader(_channel); + _logger = logger.CreateChildLogger(nameof(MqttChannelAdapter)); } public string Endpoint => _channel.Endpoint; - public MqttPacketSerializerAdapter PacketSerializerAdapter { get; } + public MqttPacketFormatterAdapter PacketFormatterAdapter { get; } public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; @@ -96,9 +98,9 @@ namespace MQTTnet.Adapter await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); try { - var packetData = PacketSerializerAdapter.Encode(packet); + var packetData = PacketFormatterAdapter.Encode(packet); await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false); - PacketSerializerAdapter.FreeBuffer(); + PacketFormatterAdapter.FreeBuffer(); _logger.Verbose("TX >>> {0}", packet); } @@ -139,12 +141,12 @@ namespace MQTTnet.Adapter return null; } - if (!PacketSerializerAdapter.ProtocolVersion.HasValue) + if (!PacketFormatterAdapter.ProtocolVersion.HasValue) { - PacketSerializerAdapter.DetectProtocolVersion(receivedMqttPacket); + PacketFormatterAdapter.DetectProtocolVersion(receivedMqttPacket); } - var packet = PacketSerializerAdapter.Decode(receivedMqttPacket); + var packet = PacketFormatterAdapter.Decode(receivedMqttPacket); if (packet == null) { throw new MqttProtocolViolationException("Received malformed packet."); @@ -172,7 +174,7 @@ namespace MQTTnet.Adapter private async Task ReceiveAsync(CancellationToken cancellationToken) { - var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(_channel, _fixedHeaderBuffer, _singleByteBuffer, cancellationToken).ConfigureAwait(false); + var fixedHeader = await _packetReader.ReadFixedHeaderAsync(_fixedHeaderBuffer, cancellationToken).ConfigureAwait(false); try { @@ -192,15 +194,15 @@ namespace MQTTnet.Adapter var bytesLeft = body.Length - bodyOffset; if (chunkSize > bytesLeft) { - chunkSize = bytesLeft; + chunkSize = (uint)bytesLeft; } #if WINDOWS_UWP - var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false); + var readBytes = await _channel.ReadAsync(body, bodyOffset, (int)chunkSize, cancellationToken).ConfigureAwait(false); #else // async/await is not used to avoid the overhead of context switches. We assume that the reamining data // has been sent from the sender directly after the initial bytes. - var readBytes = _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult(); + var readBytes = _channel.ReadAsync(body, bodyOffset, (int)chunkSize, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult(); #endif cancellationToken.ThrowIfCancellationRequested(); diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index c5c2d638..f525d2d8 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -2,11 +2,13 @@ using System.Collections.Generic; using System.Diagnostics; using System.Linq; +using System.Net; using System.Threading; using System.Threading.Tasks; using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; +using MQTTnet.Formatter; using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Protocol; @@ -156,9 +158,11 @@ namespace MQTTnet.Client public Task PublishAsync(MqttApplicationMessage applicationMessage) { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + ThrowIfNotConnected(); - var publishPacket = applicationMessage.ToPublishPacket(); + var publishPacket = _adapter.PacketFormatterAdapter.ConvertApplicationMessageToPublishPacket(applicationMessage); switch (applicationMessage.QualityOfServiceLevel) { @@ -205,6 +209,9 @@ namespace MQTTnet.Client WillMessage = willApplicationMessage }; + //connectPacket.RequestProblemInformationProperty = true; + //connectPacket.RequestResponseInformationProperty = true; + var response = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false); if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { @@ -516,6 +523,7 @@ namespace MQTTnet.Client { try { + // TODO: Move conversion to formatter. var applicationMessage = publishPacket.ToApplicationMessage(); ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); } diff --git a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs index b4a26b84..e37d2a76 100644 --- a/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs +++ b/Source/MQTTnet/Formatter/IMqttPacketFormatter.cs @@ -10,6 +10,8 @@ namespace MQTTnet.Formatter MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket); + MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage); + void FreeBuffer(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Formatter/MqttFixedHeader.cs b/Source/MQTTnet/Formatter/MqttFixedHeader.cs index c9cd3ce1..4395d875 100644 --- a/Source/MQTTnet/Formatter/MqttFixedHeader.cs +++ b/Source/MQTTnet/Formatter/MqttFixedHeader.cs @@ -2,7 +2,7 @@ { public struct MqttFixedHeader { - public MqttFixedHeader(byte flags, int remainingLength) + public MqttFixedHeader(byte flags, uint remainingLength) { Flags = flags; RemainingLength = remainingLength; @@ -10,6 +10,6 @@ public byte Flags { get; } - public int RemainingLength { get; } + public uint RemainingLength { get; } } } diff --git a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs index 3e477475..db971f09 100644 --- a/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs +++ b/Source/MQTTnet/Formatter/MqttPacketBodyReader.cs @@ -1,71 +1,140 @@ using System; +using System.Runtime.CompilerServices; using System.Text; +using MQTTnet.Exceptions; namespace MQTTnet.Formatter { public class MqttPacketBodyReader { private readonly byte[] _buffer; - private readonly int _initialOffset; - private readonly int _length; + private readonly ulong _initialOffset; + private readonly ulong _length; - private int _offset; - public MqttPacketBodyReader(byte[] buffer, int offset, int length) + : this(buffer, (ulong)offset, (ulong)length) + { + } + + public MqttPacketBodyReader(byte[] buffer, ulong offset, ulong length) { _buffer = buffer; _initialOffset = offset; - _offset = offset; + Offset = offset; _length = length; } - public int Length => _length - _offset; + public ulong Offset { get; private set; } - public bool EndOfStream => _offset == _length; + public ulong Length => _length - Offset; - public void Seek(int position) + public bool EndOfStream => Offset == _length; + + public void Seek(ulong position) { - _offset = _initialOffset + position; + Offset = _initialOffset + position; + } + + public ArraySegment Read(uint length) + { + ValidateReceiveBuffer(length); + + var buffer = new ArraySegment(_buffer, (int)Offset, (int)length); + Offset += length; + return buffer; } public byte ReadByte() { ValidateReceiveBuffer(1); - return _buffer[_offset++]; + return _buffer[Offset++]; + } + + public bool ReadBoolean() + { + ValidateReceiveBuffer(1); + var buffer = _buffer[Offset++]; + + if (buffer == 0) + { + return false; + } + + if (buffer == 1) + { + return true; + } + + throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only."); } public ArraySegment ReadRemainingData() { - return new ArraySegment(_buffer, _offset, _length - _offset); + return new ArraySegment(_buffer, (int)Offset, (int)(_length - Offset)); } - public ushort ReadUInt16() + public ushort ReadTwoByteInteger() { ValidateReceiveBuffer(2); - var msb = _buffer[_offset++]; - var lsb = _buffer[_offset++]; + var msb = _buffer[Offset++]; + var lsb = _buffer[Offset++]; return (ushort)(msb << 8 | lsb); } + public uint ReadFourByteInteger() + { + ValidateReceiveBuffer(4); + + var byte0 = _buffer[Offset++]; + var byte1 = _buffer[Offset++]; + var byte2 = _buffer[Offset++]; + var byte3 = _buffer[Offset++]; + + return (uint)(byte0 << 24 | byte1 << 16 | byte2 << 8 | byte3); + } + public ArraySegment ReadWithLengthPrefix() { - var length = ReadUInt16(); + var length = ReadTwoByteInteger(); ValidateReceiveBuffer(length); - var result = new ArraySegment(_buffer, _offset, length); - _offset += length; + var result = new ArraySegment(_buffer, (int)Offset, length); + Offset += length; return result; } - private void ValidateReceiveBuffer(ushort length) + public uint ReadVariableLengthInteger() { - if (_length < _offset + length) + var multiplier = 1; + var value = 0U; + byte encodedByte; + + do { - throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_length} bytes"); + 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; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private void ValidateReceiveBuffer(uint length) + { + if (_length < Offset + length) + { + throw new ArgumentOutOfRangeException(nameof(_buffer), $"Expected at least {Offset + length} bytes but there are only {_length} bytes"); } } diff --git a/Source/MQTTnet/Formatter/MqttPacketSerializerAdapter.cs b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs similarity index 74% rename from Source/MQTTnet/Formatter/MqttPacketSerializerAdapter.cs rename to Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs index 82bfa99c..f04467cc 100644 --- a/Source/MQTTnet/Formatter/MqttPacketSerializerAdapter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs @@ -8,44 +8,56 @@ using MQTTnet.Packets; namespace MQTTnet.Formatter { - public class MqttPacketSerializerAdapter + public class MqttPacketFormatterAdapter { - public MqttPacketSerializerAdapter() + private IMqttPacketFormatter _formatter; + + public MqttPacketFormatterAdapter() { } - public MqttPacketSerializerAdapter(MqttProtocolVersion protocolVersion) + public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion) { UseProtocolVersion(protocolVersion); } public MqttProtocolVersion? ProtocolVersion { get; private set; } - - public IMqttPacketFormatter Formatter { get; private set; } - - public ArraySegment Encode(MqttBasePacket packet) + + public MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage) { - if (Formatter == null) + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + + if (_formatter == null) { throw new InvalidOperationException("Protocol version not set or detected."); } - return Formatter.Encode(packet); + return _formatter.ConvertApplicationMessageToPublishPacket(applicationMessage); + } + + public ArraySegment Encode(MqttBasePacket packet) + { + if (_formatter == null) + { + throw new InvalidOperationException("Protocol version not set or detected."); + } + + return _formatter.Encode(packet); } public MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket) { - if (Formatter == null) + if (_formatter == null) { throw new InvalidOperationException("Protocol version not set or detected."); } - return Formatter.Decode(receivedMqttPacket); + return _formatter.Decode(receivedMqttPacket); } public void FreeBuffer() { - Formatter?.FreeBuffer(); + _formatter?.FreeBuffer(); } public void DetectProtocolVersion(ReceivedMqttPacket receivedMqttPacket) @@ -68,20 +80,20 @@ namespace MQTTnet.Formatter { case MqttProtocolVersion.V500: { - Formatter = new MqttV500PacketFormatter(); + _formatter = new MqttV500PacketFormatter(); break; } case MqttProtocolVersion.V311: { - Formatter = new MqttV311PacketFormatter(); + _formatter = new MqttV311PacketFormatter(); break; } case MqttProtocolVersion.V310: { - Formatter = new MqttV310PacketFormatter(); + _formatter = new MqttV310PacketFormatter(); break; } diff --git a/Source/MQTTnet/Formatter/MqttPacketReader.cs b/Source/MQTTnet/Formatter/MqttPacketReader.cs index 4a12efc5..97b18279 100644 --- a/Source/MQTTnet/Formatter/MqttPacketReader.cs +++ b/Source/MQTTnet/Formatter/MqttPacketReader.cs @@ -1,4 +1,5 @@ -using System.Threading; +using System; +using System.Threading; using System.Threading.Tasks; using MQTTnet.Channel; using MQTTnet.Exceptions; @@ -6,9 +7,18 @@ using MQTTnet.Internal; namespace MQTTnet.Formatter { - public static class MqttPacketReader + public class MqttPacketReader { - public static async Task ReadFixedHeaderAsync(IMqttChannel channel, byte[] fixedHeaderBuffer, byte[] singleByteBuffer, CancellationToken cancellationToken) + private readonly byte[] _singleByteBuffer = new byte[1]; + + private readonly IMqttChannel _channel; + + public MqttPacketReader(IMqttChannel channel) + { + _channel = channel ?? throw new ArgumentNullException(nameof(channel)); + } + + public async Task ReadFixedHeaderAsync(byte[] fixedHeaderBuffer, CancellationToken cancellationToken) { // The MQTT fixed header contains 1 byte of flags and at least 1 byte for the remaining data length. // So in all cases at least 2 bytes must be read for a complete MQTT packet. @@ -19,7 +29,7 @@ namespace MQTTnet.Formatter while (totalBytesRead < buffer.Length) { - var bytesRead = await channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); + var bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); ExceptionHelper.ThrowIfGracefulSocketClose(bytesRead); @@ -35,24 +45,24 @@ namespace MQTTnet.Formatter #if WINDOWS_UWP // UWP will have a dead lock when calling this not async. - var bodyLength = await ReadBodyLengthAsync(channel, buffer[1], singleByteBuffer, cancellationToken).ConfigureAwait(false); + var bodyLength = await ReadBodyLengthAsync(buffer[1], cancellationToken).ConfigureAwait(false); #else // Here the async/await pattern is not used becuase the overhead of context switches // is too big for reading 1 byte in a row. We expect that the remaining data was sent // directly after the initial bytes. If the client disconnects just in this moment we // will get an exception anyway. - var bodyLength = ReadBodyLength(channel, buffer[1], singleByteBuffer, cancellationToken); + var bodyLength = ReadBodyLength(buffer[1], cancellationToken); #endif return new MqttFixedHeader(buffer[0], bodyLength); } #if !WINDOWS_UWP - private static int ReadBodyLength(IMqttChannel channel, byte initialEncodedByte, byte[] singleByteBuffer, CancellationToken cancellationToken) + private uint ReadBodyLength(byte initialEncodedByte, CancellationToken cancellationToken) { var offset = 0; var multiplier = 128; - var value = initialEncodedByte & 127; + var value = (uint)(initialEncodedByte & 127); int encodedByte = initialEncodedByte; while ((encodedByte & 128) != 0) @@ -65,18 +75,18 @@ namespace MQTTnet.Formatter cancellationToken.ThrowIfCancellationRequested(); - encodedByte = ReadByte(channel, singleByteBuffer, cancellationToken); + encodedByte = ReadByte(cancellationToken); - value += (byte)(encodedByte & 127) * multiplier; + value += (uint)((encodedByte & 127) * multiplier); multiplier *= 128; } return value; } - private static byte ReadByte(IMqttChannel channel, byte[] singleByteBuffer, CancellationToken cancellationToken) + private byte ReadByte(CancellationToken cancellationToken) { - var readCount = channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult(); + var readCount = _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult(); cancellationToken.ThrowIfCancellationRequested(); @@ -85,16 +95,16 @@ namespace MQTTnet.Formatter ExceptionHelper.ThrowGracefulSocketClose(); } - return singleByteBuffer[0]; + return _singleByteBuffer[0]; } #else - private static async Task ReadBodyLengthAsync(IMqttChannel channel, byte initialEncodedByte, byte[] singleByteBuffer, CancellationToken cancellationToken) + private async Task ReadBodyLengthAsync(byte initialEncodedByte, CancellationToken cancellationToken) { var offset = 0; var multiplier = 128; - var value = initialEncodedByte & 127; + var value = (uint)(initialEncodedByte & 127); int encodedByte = initialEncodedByte; while ((encodedByte & 128) != 0) @@ -107,18 +117,18 @@ namespace MQTTnet.Formatter cancellationToken.ThrowIfCancellationRequested(); - encodedByte = await ReadByteAsync(channel, singleByteBuffer, cancellationToken).ConfigureAwait(false); + encodedByte = await ReadByteAsync(cancellationToken).ConfigureAwait(false); - value += (byte)(encodedByte & 127) * multiplier; + value += (uint)((encodedByte & 127) * multiplier); multiplier *= 128; } return value; } - private static async Task ReadByteAsync(IMqttChannel channel, byte[] singleByteBuffer, CancellationToken cancellationToken) + private async Task ReadByteAsync(CancellationToken cancellationToken) { - var readCount = await channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false); + var readCount = await _channel.ReadAsync(_singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); @@ -127,7 +137,7 @@ namespace MQTTnet.Formatter ExceptionHelper.ThrowGracefulSocketClose(); } - return singleByteBuffer[0]; + return _singleByteBuffer[0]; } #endif diff --git a/Source/MQTTnet/Formatter/MqttPacketWriter.cs b/Source/MQTTnet/Formatter/MqttPacketWriter.cs index 9fb8ed43..0bebfbeb 100644 --- a/Source/MQTTnet/Formatter/MqttPacketWriter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketWriter.cs @@ -17,7 +17,7 @@ namespace MQTTnet.Formatter private byte[] _buffer = new byte[128]; - private int _position; + private int _offset; public int Length { get; private set; } @@ -77,7 +77,7 @@ namespace MQTTnet.Formatter { EnsureAdditionalCapacity(1); - _buffer[_position] = @byte; + _buffer[_offset] = @byte; IncreasePostition(1); } @@ -85,9 +85,9 @@ namespace MQTTnet.Formatter { EnsureAdditionalCapacity(2); - _buffer[_position] = (byte)(value >> 8); + _buffer[_offset] = (byte)(value >> 8); IncreasePostition(1); - _buffer[_position] = (byte)value; + _buffer[_offset] = (byte)value; IncreasePostition(1); } @@ -97,7 +97,7 @@ namespace MQTTnet.Formatter EnsureAdditionalCapacity(count); - Array.Copy(buffer, offset, _buffer, _position, count); + Array.Copy(buffer, offset, _buffer, _offset, count); IncreasePostition(count); } @@ -111,7 +111,12 @@ namespace MQTTnet.Formatter public void Write(MqttPacketWriter propertyWriter) { if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter)); - + + if (propertyWriter.Length == 0) + { + return; + } + Write(propertyWriter._buffer, 0, propertyWriter.Length); } @@ -123,7 +128,7 @@ namespace MQTTnet.Formatter public void Seek(int position) { EnsureCapacity(position); - _position = position; + _offset = position; } public byte[] GetBuffer() @@ -150,7 +155,7 @@ namespace MQTTnet.Formatter [MethodImpl(MethodImplOptions.AggressiveInlining)] private void EnsureAdditionalCapacity(int additionalCapacity) { - var freeSpace = _buffer.Length - _position; + var freeSpace = _buffer.Length - _offset; if (freeSpace >= additionalCapacity) { return; @@ -180,11 +185,11 @@ namespace MQTTnet.Formatter [MethodImpl(MethodImplOptions.AggressiveInlining)] private void IncreasePostition(int length) { - _position += length; + _offset += length; - if (_position > Length) + if (_offset > Length) { - Length = _position; + Length = _offset; } } } diff --git a/Source/MQTTnet/Formatter/V310/MqttV310PacketFormatter.cs b/Source/MQTTnet/Formatter/V310/MqttV310PacketFormatter.cs index 338b506d..b1edc458 100644 --- a/Source/MQTTnet/Formatter/V310/MqttV310PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V310/MqttV310PacketFormatter.cs @@ -70,6 +70,23 @@ namespace MQTTnet.Formatter.V310 } } + public virtual MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage) + { + if (applicationMessage.UserProperties != null) + { + throw new MqttProtocolViolationException("User properties are not supported in MQTT version 3."); + } + + return new MqttPublishPacket + { + Topic = applicationMessage.Topic, + Payload = applicationMessage.Payload, + QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, + Retain = applicationMessage.Retain, + Dup = false + }; + } + public void FreeBuffer() { _packetWriter.FreeBuffer(); @@ -103,7 +120,7 @@ namespace MQTTnet.Formatter.V310 return new MqttUnsubAckPacket { - PacketIdentifier = body.ReadUInt16() + PacketIdentifier = body.ReadTwoByteInteger() }; } @@ -113,7 +130,7 @@ namespace MQTTnet.Formatter.V310 return new MqttPubCompPacket { - PacketIdentifier = body.ReadUInt16() + PacketIdentifier = body.ReadTwoByteInteger() }; } @@ -123,7 +140,7 @@ namespace MQTTnet.Formatter.V310 return new MqttPubRelPacket { - PacketIdentifier = body.ReadUInt16() + PacketIdentifier = body.ReadTwoByteInteger() }; } @@ -133,7 +150,7 @@ namespace MQTTnet.Formatter.V310 return new MqttPubRecPacket { - PacketIdentifier = body.ReadUInt16() + PacketIdentifier = body.ReadTwoByteInteger() }; } @@ -143,7 +160,7 @@ namespace MQTTnet.Formatter.V310 return new MqttPubAckPacket { - PacketIdentifier = body.ReadUInt16() + PacketIdentifier = body.ReadTwoByteInteger() }; } @@ -153,7 +170,7 @@ namespace MQTTnet.Formatter.V310 var packet = new MqttUnsubscribePacket { - PacketIdentifier = body.ReadUInt16(), + PacketIdentifier = body.ReadTwoByteInteger(), }; while (!body.EndOfStream) @@ -170,7 +187,7 @@ namespace MQTTnet.Formatter.V310 var packet = new MqttSubscribePacket { - PacketIdentifier = body.ReadUInt16() + PacketIdentifier = body.ReadTwoByteInteger() }; while (!body.EndOfStream) @@ -196,7 +213,7 @@ namespace MQTTnet.Formatter.V310 ushort? packetIdentifier = null; if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - packetIdentifier = receivedMqttPacket.Body.ReadUInt16(); + packetIdentifier = receivedMqttPacket.Body.ReadTwoByteInteger(); } var packet = new MqttPublishPacket @@ -236,7 +253,7 @@ namespace MQTTnet.Formatter.V310 var passwordFlag = (connectFlags & 0x40) > 0; var usernameFlag = (connectFlags & 0x80) > 0; - packet.KeepAlivePeriod = body.ReadUInt16(); + packet.KeepAlivePeriod = body.ReadTwoByteInteger(); packet.ClientId = body.ReadStringWithLengthPrefix(); if (willFlag) @@ -270,7 +287,7 @@ namespace MQTTnet.Formatter.V310 var packet = new MqttSubAckPacket { - PacketIdentifier = body.ReadUInt16() + PacketIdentifier = body.ReadTwoByteInteger() }; while (!body.EndOfStream) @@ -394,7 +411,7 @@ namespace MQTTnet.Formatter.V310 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); } - private static byte SerializePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter) + protected virtual byte SerializePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter) { ValidatePublishPacket(packet); @@ -439,7 +456,7 @@ namespace MQTTnet.Formatter.V310 return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader); } - private static byte SerializePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter) + protected virtual byte SerializePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { diff --git a/Source/MQTTnet/Formatter/V500/MqttV500PacketDecoder.cs b/Source/MQTTnet/Formatter/V500/MqttV500PacketDecoder.cs index f98ab7f6..348afd63 100644 --- a/Source/MQTTnet/Formatter/V500/MqttV500PacketDecoder.cs +++ b/Source/MQTTnet/Formatter/V500/MqttV500PacketDecoder.cs @@ -1,6 +1,135 @@ -namespace MQTTnet.Formatter.V500 +using System; +using System.Collections.Generic; +using System.Linq; +using MQTTnet.Exceptions; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter.V500 { public class MqttV500PacketDecoder { + public MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body) + { + if (body == null) throw new ArgumentNullException(nameof(body)); + + ThrowIfBodyIsEmpty(body); + + var packet = new MqttConnAckPacket(); + + var acknowledgeFlags = body.ReadByte(); + + packet.IsSessionPresent = (acknowledgeFlags & 0x1) > 0; + packet.ConnectReturnCode = (MqttConnectReturnCode)body.ReadByte(); + packet.ConnectReasonCode = (MqttConnectReasonCode)body.ReadByte(); + + var propertiesLength = body.ReadVariableLengthInteger(); + if (propertiesLength > 0) + { + packet.Properties = new MqttConnAckPacketProperties(); + + while (!body.EndOfStream) + { + var propertyID = (MqttPropertyID)body.ReadByte(); + + if (propertyID == MqttPropertyID.SessionExpiryInterval) + { + packet.Properties.SessionExpiryInterval = body.ReadFourByteInteger(); + } + else if (propertyID == MqttPropertyID.AuthenticationMethod) + { + packet.Properties.AuthenticationMethod = body.ReadStringWithLengthPrefix(); + } + else if (propertyID == MqttPropertyID.AuthenticationData) + { + packet.Properties.AuthenticationData = body.ReadWithLengthPrefix().ToArray(); + } + else if (propertyID == MqttPropertyID.RetainAvailable) + { + packet.Properties.RetainAvailable = body.ReadBoolean(); + } + else if (propertyID == MqttPropertyID.ReceiveMaximum) + { + packet.Properties.ReceiveMaximum = body.ReadTwoByteInteger(); + } + else if (propertyID == MqttPropertyID.AssignedClientIdentifer) + { + packet.Properties.AssignedClientIdentifier = body.ReadStringWithLengthPrefix(); + } + else if (propertyID == MqttPropertyID.TopicAliasMaximum) + { + packet.Properties.TopicAliasMaximum = body.ReadTwoByteInteger(); + } + else if (propertyID == MqttPropertyID.ReasonString) + { + packet.Properties.ReasonString = body.ReadStringWithLengthPrefix(); + } + else if (propertyID == MqttPropertyID.MaximumPacketSize) + { + packet.Properties.MaximumPacketSize = body.ReadFourByteInteger(); + } + else if (propertyID == MqttPropertyID.WildcardSubscriptionAvailable) + { + packet.Properties.WildcardSubscriptionAvailable = body.ReadBoolean(); + } + else if (propertyID == MqttPropertyID.SubscriptionIdentifierAvailable) + { + packet.Properties.SubscriptionIdentifiersAvailable = body.ReadBoolean(); + } + else if (propertyID == MqttPropertyID.SharedSubscriptionAvailable) + { + packet.Properties.SharedSubscriptionAvailable = body.ReadBoolean(); + } + else if (propertyID == MqttPropertyID.ServerKeepAlive) + { + packet.Properties.ServerKeepAlive = body.ReadTwoByteInteger(); + } + else if (propertyID == MqttPropertyID.ResponseInformation) + { + packet.Properties.ResponseInformation = body.ReadStringWithLengthPrefix(); + } + else if (propertyID == MqttPropertyID.ServerReference) + { + packet.Properties.ServerReference = body.ReadStringWithLengthPrefix(); + } + else if (propertyID == MqttPropertyID.UserProperty) + { + packet.Properties.UserProperties = ReadUserProperties(body); + } + } + } + + return packet; + } + + // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local + private static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body) + { + if (body == null || body.Length == 0) + { + throw new MqttProtocolViolationException("Data from the body is required but not present."); + } + } + + private static List ReadUserProperties(MqttPacketBodyReader reader) + { + var userPropertiesLength = reader.ReadVariableLengthInteger(); + if (userPropertiesLength == 0) + { + return new List(); + } + + var userProperties = new List(); + var targetPosition = reader.Offset + userPropertiesLength; + while (reader.Offset < targetPosition) + { + var name = reader.ReadStringWithLengthPrefix(); + var value = reader.ReadStringWithLengthPrefix(); + + userProperties.Add(new MqttUserProperty(name, value)); + } + + return userProperties; + } } } diff --git a/Source/MQTTnet/Formatter/V500/MqttV500PacketEncoder.cs b/Source/MQTTnet/Formatter/V500/MqttV500PacketEncoder.cs index f697d89f..9d236bea 100644 --- a/Source/MQTTnet/Formatter/V500/MqttV500PacketEncoder.cs +++ b/Source/MQTTnet/Formatter/V500/MqttV500PacketEncoder.cs @@ -1,6 +1,223 @@ -namespace MQTTnet.Formatter.V500 +using System; +using MQTTnet.Exceptions; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter.V500 { public class MqttV500PacketEncoder { + public byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) + { + if (packet == null) throw new ArgumentNullException(nameof(packet)); + if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); + + if (string.IsNullOrEmpty(packet.ClientId) && !packet.CleanSession) + { + throw new MqttProtocolViolationException("CleanSession must be set if ClientId is empty [MQTT-3.1.3-7]."); + } + + packetWriter.WriteWithLengthPrefix("MQTT"); + packetWriter.Write(5); // [3.1.2.2 Protocol Version] + + byte connectFlags = 0x0; + if (packet.CleanSession) + { + connectFlags |= 0x2; + } + + if (packet.WillMessage != null) + { + connectFlags |= 0x4; + connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3); + + if (packet.WillMessage.Retain) + { + connectFlags |= 0x20; + } + } + + if (packet.Password != null && packet.Username == null) + { + throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); + } + + if (packet.Password != null) + { + connectFlags |= 0x40; + } + + if (packet.Username != null) + { + connectFlags |= 0x80; + } + + packetWriter.Write(connectFlags); + packetWriter.Write(packet.KeepAlivePeriod); + + var propertiesWriter = new MqttV500PropertiesWriter(); + if (packet.Properties != null) + { + propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.WillDelayInterval, packet.Properties.WillDelayInterval); + propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.SessionExpiryInterval, packet.Properties.SessionExpiryInterval); + propertiesWriter.Write(MqttPropertyID.AuthenticationMethod, packet.Properties.AuthenticationMethod); + propertiesWriter.Write(MqttPropertyID.AuthenticationData, packet.Properties.AuthenticationData); + propertiesWriter.Write(MqttPropertyID.RequestProblemInformation, packet.Properties.RequestProblemInformation); + propertiesWriter.Write(MqttPropertyID.RequestResponseInformation, packet.Properties.RequestResponseInformation); + propertiesWriter.Write(MqttPropertyID.ReceiveMaximum, packet.Properties.ReceiveMaximum); + propertiesWriter.Write(MqttPropertyID.TopicAlias, packet.Properties.TopicAliasMaximum); + propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.MaximumPacketSize, packet.Properties.MaximumPacketSize); + propertiesWriter.WriteUserProperties(packet.Properties.UserProperties); + propertiesWriter.WriteToPacket(packetWriter); + } + + packetWriter.WriteWithLengthPrefix(packet.ClientId); + + if (packet.WillMessage != null) + { + packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic); + packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload); + } + + if (packet.Username != null) + { + packetWriter.WriteWithLengthPrefix(packet.Username); + } + + if (packet.Password != null) + { + packetWriter.WriteWithLengthPrefix(packet.Password); + } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); + } + + public byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter) + { + if (packet == null) throw new ArgumentNullException(nameof(packet)); + if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); + + if (!packet.ConnectReasonCode.HasValue) + { + throw new MqttProtocolViolationException("The ConnectReasonCode must be set for MQTT version 5."); + } + + byte connectAcknowledgeFlags = 0x0; + if (packet.IsSessionPresent) + { + connectAcknowledgeFlags |= 0x1; + } + + packetWriter.Write(connectAcknowledgeFlags); + packetWriter.Write((byte)packet.ConnectReturnCode); + packetWriter.Write((byte)packet.ConnectReasonCode.Value); + + var propertiesWriter = new MqttV500PropertiesWriter(); + if (packet.Properties != null) + { + propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.SessionExpiryInterval, packet.Properties.SessionExpiryInterval); + propertiesWriter.Write(MqttPropertyID.AuthenticationMethod, packet.Properties.AuthenticationMethod); + propertiesWriter.Write(MqttPropertyID.AuthenticationData, packet.Properties.AuthenticationData); + propertiesWriter.Write(MqttPropertyID.RetainAvailable, packet.Properties.RetainAvailable); + propertiesWriter.Write(MqttPropertyID.ReceiveMaximum, packet.Properties.ReceiveMaximum); + propertiesWriter.Write(MqttPropertyID.AssignedClientIdentifer, packet.Properties.AssignedClientIdentifier); + propertiesWriter.Write(MqttPropertyID.TopicAliasMaximum, packet.Properties.TopicAliasMaximum); + propertiesWriter.Write(MqttPropertyID.ReasonString, packet.Properties.ReasonString); + propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.MaximumPacketSize, packet.Properties.MaximumPacketSize); + propertiesWriter.Write(MqttPropertyID.WildcardSubscriptionAvailable, packet.Properties.WildcardSubscriptionAvailable); + propertiesWriter.Write(MqttPropertyID.SubscriptionIdentifierAvailable, packet.Properties.SubscriptionIdentifiersAvailable); + propertiesWriter.Write(MqttPropertyID.SharedSubscriptionAvailable, packet.Properties.SharedSubscriptionAvailable); + propertiesWriter.Write(MqttPropertyID.ServerKeepAlive, packet.Properties.ServerKeepAlive); + propertiesWriter.Write(MqttPropertyID.ResponseInformation, packet.Properties.ResponseInformation); + propertiesWriter.Write(MqttPropertyID.ServerReference, packet.Properties.ServerReference); + propertiesWriter.WriteUserProperties(packet.Properties.UserProperties); + propertiesWriter.WriteToPacket(packetWriter); + } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); + } + + public byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter) + { + if (packet == null) throw new ArgumentNullException(nameof(packet)); + if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); + + if (packet.QualityOfServiceLevel == 0 && packet.Dup) + { + throw new MqttProtocolViolationException("Dup flag must be false for QoS 0 packets [MQTT-3.3.1-2]."); + } + + packetWriter.WriteWithLengthPrefix(packet.Topic); + + if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + { + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("Publish packet has no packet identifier."); + } + + packetWriter.Write(packet.PacketIdentifier.Value); + } + else + { + if (packet.PacketIdentifier > 0) + { + throw new MqttProtocolViolationException("Packet identifier must be empty if QoS == 0 [MQTT-2.3.1-5]."); + } + } + + var propertiesWriter = new MqttV500PropertiesWriter(); + propertiesWriter.Write(MqttPropertyID.PayloadFormatIndicator, packet.Properties.PayloadFormatIndicator); + propertiesWriter.WriteAsFourByteInteger(MqttPropertyID.MessageExpiryInterval, packet.Properties.MessageExpiryInterval); + propertiesWriter.Write(MqttPropertyID.TopicAlias, packet.Properties.TopicAlias); + propertiesWriter.Write(MqttPropertyID.ResponseTopic, packet.Properties.ResponseTopic); + propertiesWriter.Write(MqttPropertyID.CorrelationData, packet.Properties.CorrelationData); + propertiesWriter.WriteAsVariableLengthInteger(MqttPropertyID.SubscriptionIdentifier, packet.Properties.SubscriptionIdentifier); + propertiesWriter.Write(MqttPropertyID.ContentType, packet.Properties.ContentType); + propertiesWriter.WriteUserProperties(packet.Properties.UserProperties); + propertiesWriter.WriteToPacket(packetWriter); + + if (packet.Payload?.Length > 0) + { + packetWriter.Write(packet.Payload, 0, packet.Payload.Length); + } + + byte fixedHeader = 0; + + if (packet.Retain) + { + fixedHeader |= 0x01; + } + + fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1); + + if (packet.Dup) + { + fixedHeader |= 0x08; + } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader); + } + + public byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter) + { + if (packet == null) throw new ArgumentNullException(nameof(packet)); + if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); + + if (!packet.PacketIdentifier.HasValue) + { + throw new MqttProtocolViolationException("PubAck packet has no packet identifier."); + } + + if (!packet.ConnectReasonCode.HasValue) + { + throw new MqttProtocolViolationException("PubAck packet must contain a connect reason."); + } + + packetWriter.Write(packet.PacketIdentifier.Value); + packetWriter.Write((byte)packet.ConnectReasonCode.Value); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck); + } } } diff --git a/Source/MQTTnet/Formatter/V500/MqttV500PacketFormatter.cs b/Source/MQTTnet/Formatter/V500/MqttV500PacketFormatter.cs index 34a93405..d95255f0 100644 --- a/Source/MQTTnet/Formatter/V500/MqttV500PacketFormatter.cs +++ b/Source/MQTTnet/Formatter/V500/MqttV500PacketFormatter.cs @@ -1,10 +1,6 @@ using System; -using System.Collections.Generic; -using MQTTnet.Exceptions; using MQTTnet.Formatter.V311; using MQTTnet.Packets; -using MQTTnet.Packets.Properties; -using MQTTnet.Protocol; namespace MQTTnet.Formatter.V500 { @@ -13,184 +9,47 @@ namespace MQTTnet.Formatter.V500 private readonly MqttV500PacketEncoder _encoder = new MqttV500PacketEncoder(); private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder(); + public override MqttPublishPacket ConvertApplicationMessageToPublishPacket(MqttApplicationMessage applicationMessage) + { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + + return new MqttPublishPacket + { + Topic = applicationMessage.Topic, + Payload = applicationMessage.Payload, + QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, + Retain = applicationMessage.Retain, + Dup = false, + Properties = new MqttPublishPacketProperties + { + UserProperties = applicationMessage.UserProperties + } + }; + } + protected override byte SerializeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) { - ValidateConnectPacket(packet); - - packetWriter.WriteWithLengthPrefix("MQTT"); - packetWriter.Write(5); // [3.1.2.2 Protocol Version] - - byte connectFlags = 0x0; - if (packet.CleanSession) - { - connectFlags |= 0x2; - } - - if (packet.WillMessage != null) - { - connectFlags |= 0x4; - connectFlags |= (byte)((byte)packet.WillMessage.QualityOfServiceLevel << 3); - - if (packet.WillMessage.Retain) - { - connectFlags |= 0x20; - } - } - - if (packet.Password != null && packet.Username == null) - { - throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); - } - - if (packet.Password != null) - { - connectFlags |= 0x40; - } - - if (packet.Username != null) - { - connectFlags |= 0x80; - } - - packetWriter.Write(connectFlags); - packetWriter.Write(packet.KeepAlivePeriod); - packetWriter.WriteWithLengthPrefix(packet.ClientId); - - if (packet.WillMessage != null) - { - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Topic); - packetWriter.WriteWithLengthPrefix(packet.WillMessage.Payload); - } - - if (packet.Username != null) - { - packetWriter.WriteWithLengthPrefix(packet.Username); - } - - if (packet.Password != null) - { - packetWriter.WriteWithLengthPrefix(packet.Password); - } - - WriteProperty(MqttMessagePropertyID.WillDelayInterval, packet.WillDelayIntervalProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.SessionExpiryInterval, packet.SessionExpiryIntervalProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.AuthenticationMethod, packet.AuthenticationMethodProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.AuthenticationData, packet.AuthenticationDataProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.RequestProblemInformation, packet.RequestProblemInformationProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.RequestResponseInformation, packet.RequestResponseInformationProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.ReceiveMaximum, packet.ReceiveMaximumProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.TopicAlias, packet.TopicAliasMaximumProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.MaximumPacketSize, packet.MaximumPacketSizeProperty, packetWriter); - WriteUserProperties(packet.UserPropertiesProperty, packetWriter); - - return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); + return _encoder.EncodeConnectPacket(packet, packetWriter); } protected override byte SerializeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter) { - byte connectAcknowledgeFlags = 0x0; - if (packet.IsSessionPresent) - { - connectAcknowledgeFlags |= 0x1; - } - - packetWriter.Write(connectAcknowledgeFlags); - packetWriter.Write((byte)packet.ConnectReturnCode); - - WriteProperty(MqttMessagePropertyID.SessionExpiryInterval, packet.SessionExpiryIntervalProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.AuthenticationMethod, packet.AuthenticationMethodProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.AuthenticationData, packet.AuthenticationDataProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.RetainAvailable, packet.RetainAvailableProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.ReceiveMaximum, packet.ReceiveMaximumProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.AssignedClientIdentifer, packet.AssignedClientIdentifierProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.TopicAlias, packet.TopicAliasMaximumProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.ReasonString, packet.ReasonStringProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.MaximumPacketSize, packet.MaximumPacketSizeProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.WildcardSubscriptionAvailable, packet.WildcardSubscriptionAvailableProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.SubscriptionIdentifierAvailable, packet.SubscriptionIdentifiersAvailableProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.SharedSubscriptionAvailable, packet.SharedSubscriptionAvailableProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.ServerKeepAlive, packet.ServerKeepAliveProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.ResponseInformation, packet.ResponseInformationProperty, packetWriter); - WriteProperty(MqttMessagePropertyID.ServerReference, packet.ServerReferenceProperty, packetWriter); - WriteUserProperties(packet.UserPropertiesProperty, packetWriter); - - return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); + return _encoder.EncodeConnAckPacket(packet, packetWriter); } - private static void WriteProperty(MqttMessagePropertyID id, bool? value, MqttPacketWriter target) + protected override MqttBasePacket DeserializeConnAckPacket(MqttPacketBodyReader body) { - if (!value.HasValue) - { - return; - } - - target.Write((byte)id); - target.Write(value.Value ? (byte)0x1 : (byte)0x0); + return _decoder.DecodeConnAckPacket(body); } - private static void WriteProperty(MqttMessagePropertyID id, ushort? value, MqttPacketWriter target) + protected override byte SerializePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter) { - if (!value.HasValue) - { - return; - } - - target.Write((byte)id); - target.Write(value.Value); + return _encoder.EncodePublishPacket(packet, packetWriter); } - private static void WriteProperty(MqttMessagePropertyID id, uint? value, MqttPacketWriter target) + protected override byte SerializePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter) { - if (!value.HasValue) - { - return; - } - - // TODO: Check if order must be reversed like for ushort. - target.Write((byte)id); - var bytes = BitConverter.GetBytes(value.Value); - target.Write(bytes, 0, bytes.Length); - } - - private static void WriteProperty(MqttMessagePropertyID id, string value, MqttPacketWriter target) - { - if (value == null) - { - return; - } - - target.Write((byte)id); - target.WriteWithLengthPrefix(value); - } - - private static void WriteProperty(MqttMessagePropertyID id, byte[] value, MqttPacketWriter target) - { - if (value == null) - { - return; - } - - target.Write((byte)id); - target.WriteWithLengthPrefix(value); - } - - private static void WriteUserProperties(List userProperties, MqttPacketWriter target) - { - if (userProperties == null || userProperties.Count == 0) - { - return; - } - - var propertyWriter = new MqttPacketWriter(); - foreach (var property in userProperties) - { - propertyWriter.WriteWithLengthPrefix(property.Name); - propertyWriter.WriteWithLengthPrefix(property.Value); - } - - target.Write((byte)MqttMessagePropertyID.UserProperty); - target.WriteVariableLengthInteger((uint)propertyWriter.Length); - target.Write(propertyWriter); + return _encoder.EncodePubAckPacket(packet, packetWriter); } } } diff --git a/Source/MQTTnet/Formatter/V500/MqttV500PropertiesWriter.cs b/Source/MQTTnet/Formatter/V500/MqttV500PropertiesWriter.cs new file mode 100644 index 00000000..1099227d --- /dev/null +++ b/Source/MQTTnet/Formatter/V500/MqttV500PropertiesWriter.cs @@ -0,0 +1,108 @@ +using System; +using System.Collections.Generic; +using MQTTnet.Packets; +using MQTTnet.Protocol; + +namespace MQTTnet.Formatter.V500 +{ + public class MqttV500PropertiesWriter + { + private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter(); + + public void Write(MqttPropertyID id, bool? value) + { + if (!value.HasValue) + { + return; + } + + _packetWriter.Write((byte)id); + _packetWriter.Write(value.Value ? (byte)0x1 : (byte)0x0); + } + + public void Write(MqttPropertyID id, ushort? value) + { + if (!value.HasValue) + { + return; + } + + _packetWriter.Write((byte)id); + _packetWriter.Write(value.Value); + } + + public void WriteAsVariableLengthInteger(MqttPropertyID id, uint? value) + { + if (!value.HasValue) + { + return; + } + + _packetWriter.Write((byte)id); + _packetWriter.WriteVariableLengthInteger(value.Value); + } + + public void WriteAsFourByteInteger(MqttPropertyID id, uint? value) + { + if (!value.HasValue) + { + return; + } + + _packetWriter.Write((byte)id); + _packetWriter.Write((byte)(value.Value >> 24)); + _packetWriter.Write((byte)(value.Value >> 16)); + _packetWriter.Write((byte)(value.Value >> 8)); + _packetWriter.Write((byte)value.Value); + } + + public void Write(MqttPropertyID id, string value) + { + if (value == null) + { + return; + } + + _packetWriter.Write((byte)id); + _packetWriter.WriteWithLengthPrefix(value); + } + + public void Write(MqttPropertyID id, byte[] value) + { + if (value == null) + { + return; + } + + _packetWriter.Write((byte)id); + _packetWriter.WriteWithLengthPrefix(value); + } + + public void WriteUserProperties(List userProperties) + { + if (userProperties == null || userProperties.Count == 0) + { + return; + } + + var propertyWriter = new MqttPacketWriter(); + foreach (var property in userProperties) + { + propertyWriter.WriteWithLengthPrefix(property.Name); + propertyWriter.WriteWithLengthPrefix(property.Value); + } + + _packetWriter.Write((byte)MqttPropertyID.UserProperty); + _packetWriter.WriteVariableLengthInteger((uint)propertyWriter.Length); + _packetWriter.Write(propertyWriter); + } + + public void WriteToPacket(MqttPacketWriter packetWriter) + { + if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter)); + + packetWriter.WriteVariableLengthInteger((uint)_packetWriter.Length); + packetWriter.Write(packetWriter); + } + } +} diff --git a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs index 9bf40964..4f88c74d 100644 --- a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs +++ b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs @@ -16,12 +16,12 @@ namespace MQTTnet.Implementations { case MqttClientTcpOptions _: { - return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketSerializerAdapter(options.ProtocolVersion), logger); + return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger); } case MqttClientWebSocketOptions webSocketOptions: { - return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializerAdapter(options.ProtocolVersion), logger); + return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger); } default: diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs index ec226b22..dd93b144 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs @@ -72,7 +72,7 @@ namespace MQTTnet.Implementations { try { - var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, _options), new MqttPacketSerializerAdapter(), _logger); + var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, _options), new MqttPacketFormatterAdapter(), _logger); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); } catch (Exception exception) diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index 8c6258e0..38036894 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -88,7 +88,7 @@ namespace MQTTnet.Implementations _socket.LocalEndPoint, _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6"); - var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializerAdapter(), _logger); + var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketFormatterAdapter(), _logger); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); } catch (ObjectDisposedException) diff --git a/Source/MQTTnet/Internal/MqttApplicationMessageExtensions.cs b/Source/MQTTnet/Internal/MqttApplicationMessageExtensions.cs index 79eb2c43..7eebda69 100644 --- a/Source/MQTTnet/Internal/MqttApplicationMessageExtensions.cs +++ b/Source/MQTTnet/Internal/MqttApplicationMessageExtensions.cs @@ -14,22 +14,5 @@ namespace MQTTnet.Internal Retain = publishPacket.Retain }; } - - public static MqttPublishPacket ToPublishPacket(this MqttApplicationMessage applicationMessage) - { - if (applicationMessage == null) - { - return null; - } - - return new MqttPublishPacket - { - Topic = applicationMessage.Topic, - Payload = applicationMessage.Payload, - QualityOfServiceLevel = applicationMessage.QualityOfServiceLevel, - Retain = applicationMessage.Retain, - Dup = false - }; - } } } diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 8b4038d0..91e455f7 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -69,6 +69,10 @@ + + + + C:\Program Files\dotnet\sdk\NuGetFallbackFolder\microsoft.netcore.app\2.1.0\ref\netcoreapp2.1\System.Net.Requests.dll diff --git a/Source/MQTTnet/MqttApplicationMessage.cs b/Source/MQTTnet/MqttApplicationMessage.cs index 334dfb40..54135fb0 100644 --- a/Source/MQTTnet/MqttApplicationMessage.cs +++ b/Source/MQTTnet/MqttApplicationMessage.cs @@ -1,4 +1,6 @@ -using MQTTnet.Protocol; +using System.Collections.Generic; +using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet { @@ -10,6 +12,8 @@ namespace MQTTnet public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; } + public List UserProperties { get; set; } + public bool Retain { get; set; } } } diff --git a/Source/MQTTnet/Packets/MqttAuthPacket.cs b/Source/MQTTnet/Packets/MqttAuthPacket.cs index aa8e9a60..191ec5c1 100644 --- a/Source/MQTTnet/Packets/MqttAuthPacket.cs +++ b/Source/MQTTnet/Packets/MqttAuthPacket.cs @@ -1,6 +1,4 @@ -using System.Collections.Generic; -using MQTTnet.Packets.Properties; -using MQTTnet.Protocol; +using MQTTnet.Protocol; namespace MQTTnet.Packets { @@ -10,7 +8,5 @@ namespace MQTTnet.Packets public class MqttAuthPacket : MqttBasePacket { public MqttAuthenticateReasonCode ReasonCode { get; set; } - - public List Properties { get; set; } } } diff --git a/Source/MQTTnet/Packets/MqttBasePublishPacket.cs b/Source/MQTTnet/Packets/MqttBasePublishPacket.cs index 67e68879..ffafc537 100644 --- a/Source/MQTTnet/Packets/MqttBasePublishPacket.cs +++ b/Source/MQTTnet/Packets/MqttBasePublishPacket.cs @@ -1,15 +1,7 @@ -using System.Collections.Generic; -using MQTTnet.Packets.Properties; - -namespace MQTTnet.Packets +namespace MQTTnet.Packets { public class MqttBasePublishPacket : MqttBasePacket, IMqttPacketWithIdentifier { public ushort? PacketIdentifier { get; set; } - - /// - /// Added in MQTTv5.0.0. - /// - public List Properties { get; set; } } } diff --git a/Source/MQTTnet/Packets/MqttConnAckPacket.cs b/Source/MQTTnet/Packets/MqttConnAckPacket.cs index d79b44af..1a9aa669 100644 --- a/Source/MQTTnet/Packets/MqttConnAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnAckPacket.cs @@ -1,5 +1,4 @@ -using System.Collections.Generic; -using MQTTnet.Protocol; +using MQTTnet.Protocol; namespace MQTTnet.Packets { @@ -14,38 +13,9 @@ namespace MQTTnet.Packets #region Added in MQTTv5 - public uint? SessionExpiryIntervalProperty { get; set; } + public MqttConnectReasonCode? ConnectReasonCode { get; set; } - // TODO: Add enum - public byte? ReceiveMaximumProperty { get; set; } - - public bool? RetainAvailableProperty { get; set; } - - public uint? MaximumPacketSizeProperty { get; set; } - - public string AssignedClientIdentifierProperty { get; set; } - - public ushort? TopicAliasMaximumProperty { get; set; } - - public string ReasonStringProperty { get; set; } - - public List UserPropertiesProperty { get; set; } - - public bool? WildcardSubscriptionAvailableProperty { get; set; } - - public bool? SubscriptionIdentifiersAvailableProperty { get; set; } - - public bool? SharedSubscriptionAvailableProperty { get; set; } - - public ushort? ServerKeepAliveProperty { get; set; } - - public string ResponseInformationProperty { get; set; } - - public string ServerReferenceProperty { get; set; } - - public string AuthenticationMethodProperty { get; set; } - - public byte[] AuthenticationDataProperty { get; set; } + public MqttConnAckPacketProperties Properties { get; set; } #endregion diff --git a/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs new file mode 100644 index 00000000..31a6b130 --- /dev/null +++ b/Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs @@ -0,0 +1,40 @@ +using System.Collections.Generic; + +namespace MQTTnet.Packets +{ + public class MqttConnAckPacketProperties + { + public uint? SessionExpiryInterval { get; set; } + + // TODO: Add enum + public ushort? ReceiveMaximum { get; set; } + + public bool? RetainAvailable { get; set; } + + public uint? MaximumPacketSize { get; set; } + + public string AssignedClientIdentifier { get; set; } + + public ushort? TopicAliasMaximum { get; set; } + + public string ReasonString { get; set; } + + public List UserProperties { get; set; } + + public bool? WildcardSubscriptionAvailable { get; set; } + + public bool? SubscriptionIdentifiersAvailable { get; set; } + + public bool? SharedSubscriptionAvailable { get; set; } + + public ushort? ServerKeepAlive { get; set; } + + public string ResponseInformation { get; set; } + + public string ServerReference { get; set; } + + public string AuthenticationMethod { get; set; } + + public byte[] AuthenticationData { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Packets/MqttConnectPacket.cs b/Source/MQTTnet/Packets/MqttConnectPacket.cs index 0708123b..1fc5a42e 100644 --- a/Source/MQTTnet/Packets/MqttConnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnectPacket.cs @@ -1,12 +1,10 @@ -using System.Collections.Generic; - -namespace MQTTnet.Packets +namespace MQTTnet.Packets { public class MqttConnectPacket : MqttBasePacket { public string ProtocolName { get; set; } - public byte ProtocolLevel { get; set; } + public byte? ProtocolLevel { get; set; } public string ClientId { get; set; } @@ -25,25 +23,7 @@ namespace MQTTnet.Packets #region Added in MQTTv5 - public uint? WillDelayIntervalProperty { get; set; } - - public uint? SessionExpiryIntervalProperty { get; set; } - - public string AuthenticationMethodProperty { get; set; } - - public byte[] AuthenticationDataProperty { get; set; } - - public bool? RequestProblemInformationProperty { get; set; } - - public bool? RequestResponseInformationProperty { get; set; } - - public ushort? ReceiveMaximumProperty { get; set; } - - public ushort? TopicAliasMaximumProperty { get; set; } - - public uint? MaximumPacketSizeProperty { get; set; } - - public List UserPropertiesProperty { get; set; } + public MqttConnectPacketProperties Properties { get; set; } #endregion diff --git a/Source/MQTTnet/Packets/MqttConnectPacketProperties.cs b/Source/MQTTnet/Packets/MqttConnectPacketProperties.cs new file mode 100644 index 00000000..4cc405e2 --- /dev/null +++ b/Source/MQTTnet/Packets/MqttConnectPacketProperties.cs @@ -0,0 +1,27 @@ +using System.Collections.Generic; + +namespace MQTTnet.Packets +{ + public class MqttConnectPacketProperties + { + public uint? WillDelayInterval { get; set; } + + public uint? SessionExpiryInterval { get; set; } + + public string AuthenticationMethod { get; set; } + + public byte[] AuthenticationData { get; set; } + + public bool? RequestProblemInformation { get; set; } + + public bool? RequestResponseInformation { get; set; } + + public ushort? ReceiveMaximum { get; set; } + + public ushort? TopicAliasMaximum { get; set; } + + public uint? MaximumPacketSize { get; set; } + + public List UserProperties { get; set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Packets/MqttDisconnectPacket.cs b/Source/MQTTnet/Packets/MqttDisconnectPacket.cs index 5dbbd497..621829a2 100644 --- a/Source/MQTTnet/Packets/MqttDisconnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttDisconnectPacket.cs @@ -1,15 +1,7 @@ -using System.Collections.Generic; -using MQTTnet.Packets.Properties; - -namespace MQTTnet.Packets +namespace MQTTnet.Packets { public class MqttDisconnectPacket : MqttBasePacket { - /// - /// Added in MQTTv5.0.0. - /// - public List Properties { get; set; } - public override string ToString() { return "Disconnect"; diff --git a/Source/MQTTnet/Packets/MqttPubAckPacket.cs b/Source/MQTTnet/Packets/MqttPubAckPacket.cs index 5811849b..ad29f079 100644 --- a/Source/MQTTnet/Packets/MqttPubAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttPubAckPacket.cs @@ -1,10 +1,20 @@ -namespace MQTTnet.Packets +using MQTTnet.Protocol; + +namespace MQTTnet.Packets { public class MqttPubAckPacket : MqttBasePublishPacket { + #region Added in MQTTv5 + + public MqttPubAckReasonCode? ConnectReasonCode { get; set; } + + public MqttPubAckPacketProperties Properties { get; set; } + + #endregion + public override string ToString() { - return $"PubAck [PacketIdentifier={PacketIdentifier}]"; + return $"PubAck: [PacketIdentifier={PacketIdentifier}] [ConnectReasonCode={ConnectReasonCode}]"; } } } diff --git a/Source/MQTTnet/Packets/MqttPubAckPacketProperties.cs b/Source/MQTTnet/Packets/MqttPubAckPacketProperties.cs new file mode 100644 index 00000000..e2debd4a --- /dev/null +++ b/Source/MQTTnet/Packets/MqttPubAckPacketProperties.cs @@ -0,0 +1,11 @@ +using System.Collections.Generic; + +namespace MQTTnet.Packets +{ + public class MqttPubAckPacketProperties + { + public string ReasonString { get; set; } + + public List UserProperties { get; set; } + } +} diff --git a/Source/MQTTnet/Packets/MqttPublishPacket.cs b/Source/MQTTnet/Packets/MqttPublishPacket.cs index 374f913b..e1644d0c 100644 --- a/Source/MQTTnet/Packets/MqttPublishPacket.cs +++ b/Source/MQTTnet/Packets/MqttPublishPacket.cs @@ -1,6 +1,4 @@ -using System.Collections.Generic; -using MQTTnet.Packets.Properties; -using MQTTnet.Protocol; +using MQTTnet.Protocol; namespace MQTTnet.Packets { @@ -16,10 +14,11 @@ namespace MQTTnet.Packets public byte[] Payload { get; set; } - /// - /// Added in MQTTv5.0.0. - /// - public List Properties { get; set; } + #region Added in MQTTv5 + + public MqttPublishPacketProperties Properties { get; set; } + + #endregion public override string ToString() { diff --git a/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs new file mode 100644 index 00000000..ba7254da --- /dev/null +++ b/Source/MQTTnet/Packets/MqttPublishPacketProperties.cs @@ -0,0 +1,23 @@ +using System.Collections.Generic; + +namespace MQTTnet.Packets +{ + public class MqttPublishPacketProperties + { + public byte? PayloadFormatIndicator { get; set; } + + public uint? MessageExpiryInterval { get; set; } + + public ushort? TopicAlias { get; set; } + + public string ResponseTopic { get; set; } + + public byte[] CorrelationData { get; set; } + + public List UserProperties { get; set; } + + public uint? SubscriptionIdentifier { get; set; } + + public string ContentType { get; set; } + } +} diff --git a/Source/MQTTnet/Packets/MqttSubscribePacket.cs b/Source/MQTTnet/Packets/MqttSubscribePacket.cs index 89a1efe6..60086c47 100644 --- a/Source/MQTTnet/Packets/MqttSubscribePacket.cs +++ b/Source/MQTTnet/Packets/MqttSubscribePacket.cs @@ -1,6 +1,5 @@ using System.Collections.Generic; using System.Linq; -using MQTTnet.Packets.Properties; namespace MQTTnet.Packets { @@ -10,11 +9,6 @@ namespace MQTTnet.Packets public IList TopicFilters { get; set; } = new List(); - /// - /// Added in MQTTv5.0.0. - /// - public List Properties { get; set; } - public override string ToString() { var topicFiltersText = string.Join(",", TopicFilters.Select(f => f.Topic + "@" + f.QualityOfServiceLevel)); diff --git a/Source/MQTTnet/Packets/Properties/AssignedClientIdentifierProperty.cs b/Source/MQTTnet/Packets/Properties/AssignedClientIdentifierProperty.cs deleted file mode 100644 index bda8e20e..00000000 --- a/Source/MQTTnet/Packets/Properties/AssignedClientIdentifierProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class AssignedClientIdentifierProperty : StringProperty - { - public AssignedClientIdentifierProperty(string value) - : base((byte)MqttMessagePropertyID.AssignedClientIdentifer, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/AuthenticationDataProperty.cs b/Source/MQTTnet/Packets/Properties/AuthenticationDataProperty.cs deleted file mode 100644 index 48cba80e..00000000 --- a/Source/MQTTnet/Packets/Properties/AuthenticationDataProperty.cs +++ /dev/null @@ -1,13 +0,0 @@ -using System; -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class AuthenticationDataProperty : BinaryDataProperty - { - public AuthenticationDataProperty(ArraySegment data) - : base((byte)MqttMessagePropertyID.AuthenticationData, data) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/AuthenticationMethodProperty.cs b/Source/MQTTnet/Packets/Properties/AuthenticationMethodProperty.cs deleted file mode 100644 index 9693278b..00000000 --- a/Source/MQTTnet/Packets/Properties/AuthenticationMethodProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class AuthenticationMethodProperty : StringProperty - { - public AuthenticationMethodProperty(string value) - : base((byte)MqttMessagePropertyID.AuthenticationMethod, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/BaseTypes/BinaryDataProperty.cs b/Source/MQTTnet/Packets/Properties/BaseTypes/BinaryDataProperty.cs deleted file mode 100644 index 484d5ee7..00000000 --- a/Source/MQTTnet/Packets/Properties/BaseTypes/BinaryDataProperty.cs +++ /dev/null @@ -1,32 +0,0 @@ -using System; -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties.BaseTypes -{ - public class BinaryDataProperty : IProperty - { - public BinaryDataProperty(byte id, ArraySegment data) - { - if (data.Array == null) - { - throw new ArgumentNullException(nameof(data)); - } - - Id = id; - Data = data; - } - - public byte Id { get; } - - public ArraySegment Data { get; } - - public void WriteTo(MqttPacketWriter writer) - { - if (writer == null) throw new ArgumentNullException(nameof(writer)); - - writer.Write(Id); - writer.Write((ushort)Data.Count); - writer.Write(Data); - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/BaseTypes/ByteProperty.cs b/Source/MQTTnet/Packets/Properties/BaseTypes/ByteProperty.cs deleted file mode 100644 index 1bfb9aab..00000000 --- a/Source/MQTTnet/Packets/Properties/BaseTypes/ByteProperty.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties.BaseTypes -{ - public class ByteProperty : IProperty - { - public ByteProperty(byte id, byte value) - { - Id = id; - Value = value; - } - - public byte Id { get; } - - public byte Value { get; } - - public void WriteTo(MqttPacketWriter writer) - { - if (writer == null) throw new ArgumentNullException(nameof(writer)); - - writer.Write(Id); - writer.Write(Value); - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/BaseTypes/FourByteIntegerProperty.cs b/Source/MQTTnet/Packets/Properties/BaseTypes/FourByteIntegerProperty.cs deleted file mode 100644 index 594ec3e0..00000000 --- a/Source/MQTTnet/Packets/Properties/BaseTypes/FourByteIntegerProperty.cs +++ /dev/null @@ -1,28 +0,0 @@ -using System; -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties.BaseTypes -{ - public class FourByteIntegerValue : IProperty - { - public FourByteIntegerValue(byte id, uint value) - { - Id = id; - Value = value; - } - - public byte Id { get; } - - public uint Value { get; } - - public void WriteTo(MqttPacketWriter writer) - { - if (writer == null) throw new ArgumentNullException(nameof(writer)); - - // TODO: Check if order must be reversed like for ushort. - writer.Write(Id); - var bytes = BitConverter.GetBytes(Value); - writer.Write(bytes, 0, bytes.Length); - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/BaseTypes/StringProperty.cs b/Source/MQTTnet/Packets/Properties/BaseTypes/StringProperty.cs deleted file mode 100644 index ff3c1fbf..00000000 --- a/Source/MQTTnet/Packets/Properties/BaseTypes/StringProperty.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties.BaseTypes -{ - public class StringProperty : IProperty - { - public StringProperty(byte id, string value) - { - Id = id; - Value = value; - } - - public byte Id { get; } - - public string Value { get; } - - public void WriteTo(MqttPacketWriter writer) - { - if (writer == null) throw new ArgumentNullException(nameof(writer)); - - writer.Write(Id); - writer.WriteWithLengthPrefix(Value); - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/BaseTypes/TwoByteIntegerProperty.cs b/Source/MQTTnet/Packets/Properties/BaseTypes/TwoByteIntegerProperty.cs deleted file mode 100644 index 8e5a4da0..00000000 --- a/Source/MQTTnet/Packets/Properties/BaseTypes/TwoByteIntegerProperty.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties.BaseTypes -{ - public class TwoByteIntegerProperty : IProperty - { - public TwoByteIntegerProperty(byte id, ushort value) - { - Id = id; - Value = value; - } - - public byte Id { get; } - - public ushort Value { get; } - - public void WriteTo(MqttPacketWriter writer) - { - if (writer == null) throw new ArgumentNullException(nameof(writer)); - - writer.Write(Id); - writer.Write(Value); - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/BaseTypes/VariableByteIntegerProperty.cs b/Source/MQTTnet/Packets/Properties/BaseTypes/VariableByteIntegerProperty.cs deleted file mode 100644 index e1fd355c..00000000 --- a/Source/MQTTnet/Packets/Properties/BaseTypes/VariableByteIntegerProperty.cs +++ /dev/null @@ -1,28 +0,0 @@ -using System; -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties.BaseTypes -{ - public class VariableByteIntegerProperty : IProperty - { - public VariableByteIntegerProperty(byte id, uint value) - { - Id = id; - Value = value; - } - - public byte Id { get; } - - public uint Value { get; } - - public void WriteTo(MqttPacketWriter writer) - { - if (writer == null) throw new ArgumentNullException(nameof(writer)); - - var buffer = MqttPacketWriter.EncodeVariableByteInteger(Value); - - writer.Write(Id); - writer.Write(buffer); - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/ContentTypeProperty.cs b/Source/MQTTnet/Packets/Properties/ContentTypeProperty.cs deleted file mode 100644 index 459b70d6..00000000 --- a/Source/MQTTnet/Packets/Properties/ContentTypeProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class ContentTypeProperty : StringProperty - { - public ContentTypeProperty(string value) - : base((byte)MqttMessagePropertyID.ContentType, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/CorrelationDataProperty.cs b/Source/MQTTnet/Packets/Properties/CorrelationDataProperty.cs deleted file mode 100644 index 8b517e3c..00000000 --- a/Source/MQTTnet/Packets/Properties/CorrelationDataProperty.cs +++ /dev/null @@ -1,13 +0,0 @@ -using System; -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class CorrelationDataProperty : BinaryDataProperty - { - public CorrelationDataProperty(ArraySegment data) - : base((byte)MqttMessagePropertyID.CorrelationData, data) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/IProperty.cs b/Source/MQTTnet/Packets/Properties/IProperty.cs deleted file mode 100644 index 484e48ba..00000000 --- a/Source/MQTTnet/Packets/Properties/IProperty.cs +++ /dev/null @@ -1,11 +0,0 @@ -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties -{ - public interface IProperty - { - byte Id { get; } - - void WriteTo(MqttPacketWriter writer); - } -} diff --git a/Source/MQTTnet/Packets/Properties/MaximumPacketSizeProperty.cs b/Source/MQTTnet/Packets/Properties/MaximumPacketSizeProperty.cs deleted file mode 100644 index be060648..00000000 --- a/Source/MQTTnet/Packets/Properties/MaximumPacketSizeProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class MaximumPacketSizeProperty : FourByteIntegerValue - { - public MaximumPacketSizeProperty(uint value) - : base((byte)MqttMessagePropertyID.MaximumPacketSize, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/MaximumQoSProperty.cs b/Source/MQTTnet/Packets/Properties/MaximumQoSProperty.cs deleted file mode 100644 index 168b8ca0..00000000 --- a/Source/MQTTnet/Packets/Properties/MaximumQoSProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class MaximumQoSProperty : ByteProperty - { - public MaximumQoSProperty(byte value) - : base((byte)MqttMessagePropertyID.MaximumQoS, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/MessageExpiryIntervalProperty.cs b/Source/MQTTnet/Packets/Properties/MessageExpiryIntervalProperty.cs deleted file mode 100644 index 6b944f04..00000000 --- a/Source/MQTTnet/Packets/Properties/MessageExpiryIntervalProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class MessageExpiryIntervalProperty : FourByteIntegerValue - { - public MessageExpiryIntervalProperty(uint value) - : base((byte)MqttMessagePropertyID.MessageExpiryInterval, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/PayloadFormatIndicatorProperty.cs b/Source/MQTTnet/Packets/Properties/PayloadFormatIndicatorProperty.cs deleted file mode 100644 index 79a7d22d..00000000 --- a/Source/MQTTnet/Packets/Properties/PayloadFormatIndicatorProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class PayloadFormatIndicatorProperty : ByteProperty - { - public PayloadFormatIndicatorProperty(byte value) - : base((byte)MqttMessagePropertyID.PayloadFormatIndicator, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/ReasonStringProperty.cs b/Source/MQTTnet/Packets/Properties/ReasonStringProperty.cs deleted file mode 100644 index 1a3d4f15..00000000 --- a/Source/MQTTnet/Packets/Properties/ReasonStringProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class ReasonStringProperty : StringProperty - { - public ReasonStringProperty(string value) - : base((byte)MqttMessagePropertyID.ReasonString, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/ReceiveMaximumProperty.cs b/Source/MQTTnet/Packets/Properties/ReceiveMaximumProperty.cs deleted file mode 100644 index 3263eb03..00000000 --- a/Source/MQTTnet/Packets/Properties/ReceiveMaximumProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class ReceiveMaximumProperty : TwoByteIntegerProperty - { - public ReceiveMaximumProperty(ushort value) - : base((byte)MqttMessagePropertyID.ReceiveMaximum, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/RequestProblemInformationProperty.cs b/Source/MQTTnet/Packets/Properties/RequestProblemInformationProperty.cs deleted file mode 100644 index 537257c6..00000000 --- a/Source/MQTTnet/Packets/Properties/RequestProblemInformationProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class RequestProblemInformationProperty : ByteProperty - { - public RequestProblemInformationProperty(byte value) - : base((byte)MqttMessagePropertyID.RequestProblemInformation, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/RequestResponseInformationProperty.cs b/Source/MQTTnet/Packets/Properties/RequestResponseInformationProperty.cs deleted file mode 100644 index 881f7eae..00000000 --- a/Source/MQTTnet/Packets/Properties/RequestResponseInformationProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class RequestResponseInformationProperty : ByteProperty - { - public RequestResponseInformationProperty(byte value) - : base((byte)MqttMessagePropertyID.RequestResponseInformation, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/ResponseInformationProperty.cs b/Source/MQTTnet/Packets/Properties/ResponseInformationProperty.cs deleted file mode 100644 index e080f992..00000000 --- a/Source/MQTTnet/Packets/Properties/ResponseInformationProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class ResponseInformationProperty : StringProperty - { - public ResponseInformationProperty(string value) - : base((byte)MqttMessagePropertyID.ResponseInformation, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/ResponseTopicProperty.cs b/Source/MQTTnet/Packets/Properties/ResponseTopicProperty.cs deleted file mode 100644 index 5f221863..00000000 --- a/Source/MQTTnet/Packets/Properties/ResponseTopicProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class ResponseTopicProperty : StringProperty - { - public ResponseTopicProperty(string value) - : base((byte)MqttMessagePropertyID.ResponseTopic, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/RetainAvailableProperty.cs b/Source/MQTTnet/Packets/Properties/RetainAvailableProperty.cs deleted file mode 100644 index f407bdcc..00000000 --- a/Source/MQTTnet/Packets/Properties/RetainAvailableProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class RetainAvailableProperty : ByteProperty - { - public RetainAvailableProperty(byte value) - : base((byte)MqttMessagePropertyID.RetainAvailable, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/ServerKeepAliveProperty.cs b/Source/MQTTnet/Packets/Properties/ServerKeepAliveProperty.cs deleted file mode 100644 index 272860c4..00000000 --- a/Source/MQTTnet/Packets/Properties/ServerKeepAliveProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class ServerKeepAliveProperty : TwoByteIntegerProperty - { - public ServerKeepAliveProperty(ushort value) - : base((byte)MqttMessagePropertyID.ServerKeepAlive, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/ServerReferenceProperty.cs b/Source/MQTTnet/Packets/Properties/ServerReferenceProperty.cs deleted file mode 100644 index 2b50e793..00000000 --- a/Source/MQTTnet/Packets/Properties/ServerReferenceProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class ServerReferenceProperty : StringProperty - { - public ServerReferenceProperty(string value) - : base((byte)MqttMessagePropertyID.ServerReference, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/SessionExpiryIntervalProperty.cs b/Source/MQTTnet/Packets/Properties/SessionExpiryIntervalProperty.cs deleted file mode 100644 index 651de5d2..00000000 --- a/Source/MQTTnet/Packets/Properties/SessionExpiryIntervalProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class SessionExpiryIntervalProperty : FourByteIntegerValue - { - public SessionExpiryIntervalProperty(uint value) - : base((byte)MqttMessagePropertyID.SessionExpiryInterval, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/SharedSubscriptionAvailableProperty.cs b/Source/MQTTnet/Packets/Properties/SharedSubscriptionAvailableProperty.cs deleted file mode 100644 index 8461fa38..00000000 --- a/Source/MQTTnet/Packets/Properties/SharedSubscriptionAvailableProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class SharedSubscriptionAvailableProperty : ByteProperty - { - public SharedSubscriptionAvailableProperty(byte value) - : base((byte)MqttMessagePropertyID.SharedSubscriptionAvailable, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/SubscriptionIdentifierAvailable.cs b/Source/MQTTnet/Packets/Properties/SubscriptionIdentifierAvailable.cs deleted file mode 100644 index d4789b23..00000000 --- a/Source/MQTTnet/Packets/Properties/SubscriptionIdentifierAvailable.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class SubscriptionIdentifierAvailable : ByteProperty - { - public SubscriptionIdentifierAvailable(byte value) - : base((byte)MqttMessagePropertyID.SubscriptionIdentifierAvailable, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/SubscriptionIdentifierProperty.cs b/Source/MQTTnet/Packets/Properties/SubscriptionIdentifierProperty.cs deleted file mode 100644 index 9f9a56fe..00000000 --- a/Source/MQTTnet/Packets/Properties/SubscriptionIdentifierProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class SubscriptionIdentifierProperty : VariableByteIntegerProperty - { - public SubscriptionIdentifierProperty(uint value) - : base((byte)MqttMessagePropertyID.SubscriptionIdentifier, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/TopicAliasMaximumProperty.cs b/Source/MQTTnet/Packets/Properties/TopicAliasMaximumProperty.cs deleted file mode 100644 index 02c36c08..00000000 --- a/Source/MQTTnet/Packets/Properties/TopicAliasMaximumProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class TopicAliasMaximumProperty : TwoByteIntegerProperty - { - public TopicAliasMaximumProperty(ushort value) - : base((byte)MqttMessagePropertyID.TopicAliasMaximum, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/TopicAliasProperty.cs b/Source/MQTTnet/Packets/Properties/TopicAliasProperty.cs deleted file mode 100644 index 41c4495e..00000000 --- a/Source/MQTTnet/Packets/Properties/TopicAliasProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class TopicAliasProperty : TwoByteIntegerProperty - { - public TopicAliasProperty(ushort value) - : base((byte)MqttMessagePropertyID.TopicAlias, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/UserPropertyProperty.cs b/Source/MQTTnet/Packets/Properties/UserPropertyProperty.cs deleted file mode 100644 index 0708d3f4..00000000 --- a/Source/MQTTnet/Packets/Properties/UserPropertyProperty.cs +++ /dev/null @@ -1,29 +0,0 @@ -using System; -using MQTTnet.Formatter; - -namespace MQTTnet.Packets.Properties -{ - public class UserPropertyProperty : IProperty - { - public UserPropertyProperty(string name, string value) - { - Name = name ?? throw new ArgumentNullException(nameof(name)); - Value = value ?? throw new ArgumentNullException(nameof(value)); - } - - public byte Id { get; } = (byte)MqttMessagePropertyID.UserProperty; - - public string Name { get; } - - public string Value { get; } - - public void WriteTo(MqttPacketWriter writer) - { - if (writer == null) throw new ArgumentNullException(nameof(writer)); - - writer.Write(Id); - writer.WriteWithLengthPrefix(Name); - writer.WriteWithLengthPrefix(Value); - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/WildcardSubscriptionAvailableProperty.cs b/Source/MQTTnet/Packets/Properties/WildcardSubscriptionAvailableProperty.cs deleted file mode 100644 index 7c901ec4..00000000 --- a/Source/MQTTnet/Packets/Properties/WildcardSubscriptionAvailableProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class WildcardSubscriptionAvailableProperty : ByteProperty - { - public WildcardSubscriptionAvailableProperty(byte value) - : base((byte)MqttMessagePropertyID.WildcardSubscriptionAvailable, value) - { - } - } -} diff --git a/Source/MQTTnet/Packets/Properties/WillDelayIntervalProperty.cs b/Source/MQTTnet/Packets/Properties/WillDelayIntervalProperty.cs deleted file mode 100644 index f40bfbc3..00000000 --- a/Source/MQTTnet/Packets/Properties/WillDelayIntervalProperty.cs +++ /dev/null @@ -1,12 +0,0 @@ -using MQTTnet.Packets.Properties.BaseTypes; - -namespace MQTTnet.Packets.Properties -{ - public class WillDelayIntervalProperty : FourByteIntegerValue - { - public WillDelayIntervalProperty(uint value) - : base((byte)MqttMessagePropertyID.WillDelayInterval, value) - { - } - } -} diff --git a/Source/MQTTnet/Protocol/MqttConnectReasonCode.cs b/Source/MQTTnet/Protocol/MqttConnectReasonCode.cs new file mode 100644 index 00000000..381c6fa5 --- /dev/null +++ b/Source/MQTTnet/Protocol/MqttConnectReasonCode.cs @@ -0,0 +1,28 @@ +namespace MQTTnet.Protocol +{ + public enum MqttConnectReasonCode + { + Success = 0, + UnspecifiedError = 128, + MalformedPacket = 129, + ProtocolError = 130, + ImplementationSpecificError = 131, + UnsupportedProtocolVersion = 132, + ClientIdentifierNotValid = 133, + BadUserNameorPassword = 134, + NotAuthorized = 135, + ServerUnavailable = 136, + ServerBusy = 137, + Banned = 138, + BadAuthenticationMethod = 140, + TopicNameInvalid = 144, + PacketTooLarge = 149, + QuotaExceeded = 151, + PayloadFormatInvalid = 153, + RetainNotSupported = 154, + QoSNotSupported = 155, + UseAnotherServer = 156, + ServerMoved = 157, + ConnectionRateExceeded = 159 + } +} diff --git a/Source/MQTTnet/Packets/Properties/MqttMessagePropertyID.cs b/Source/MQTTnet/Protocol/MqttPropertyID.cs similarity index 92% rename from Source/MQTTnet/Packets/Properties/MqttMessagePropertyID.cs rename to Source/MQTTnet/Protocol/MqttPropertyID.cs index 2b8666d9..704d8dfc 100644 --- a/Source/MQTTnet/Packets/Properties/MqttMessagePropertyID.cs +++ b/Source/MQTTnet/Protocol/MqttPropertyID.cs @@ -1,6 +1,6 @@ -namespace MQTTnet.Packets.Properties +namespace MQTTnet.Protocol { - public enum MqttMessagePropertyID + public enum MqttPropertyID { PayloadFormatIndicator = 1, MessageExpiryInterval = 2, diff --git a/Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs b/Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs new file mode 100644 index 00000000..b0580d6a --- /dev/null +++ b/Source/MQTTnet/Protocol/MqttPubAckReasonCode.cs @@ -0,0 +1,15 @@ +namespace MQTTnet.Protocol +{ + public enum MqttPubAckReasonCode + { + Success = 0, + NoMatchingSubscribers = 16, + UnspecifiedError = 128, + ImplementationSpecificError = 131, + NotAuthorized = 135, + TopicNameInvalid = 144, + PacketIdentifierInUse = 145, + QuotaExceeded = 151, + PayloadFormatInvalid = 153 + } +} diff --git a/Source/MQTTnet/Server/IMqttClientSession.cs b/Source/MQTTnet/Server/IMqttClientSession.cs index e0dd214c..6e29a2f2 100644 --- a/Source/MQTTnet/Server/IMqttClientSession.cs +++ b/Source/MQTTnet/Server/IMqttClientSession.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Threading.Tasks; using MQTTnet.Adapter; -using MQTTnet.Packets; namespace MQTTnet.Server { @@ -11,7 +10,7 @@ namespace MQTTnet.Server string ClientId { get; } void FillStatus(MqttClientSessionStatus status); - void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket); + void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage); void ClearPendingApplicationMessages(); Task RunAsync(MqttApplicationMessage willMessage, int keepAliveInterval, IMqttChannelAdapter adapter); diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 1005110b..cc949a48 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -61,7 +61,7 @@ namespace MQTTnet.Server status.ClientId = ClientId; status.IsConnected = _adapter != null; status.Endpoint = _adapter?.Endpoint; - status.ProtocolVersion = _adapter?.PacketSerializerAdapter?.ProtocolVersion; + status.ProtocolVersion = _adapter?.PacketFormatterAdapter?.ProtocolVersion; status.PendingApplicationMessagesCount = _pendingPacketsQueue.Count; status.LastPacketReceived = _keepAliveMonitor.LastPacketReceived; status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; @@ -193,7 +193,7 @@ namespace MQTTnet.Server if (_willMessage != null && !_wasCleanDisconnect) { - _sessionsManager.EnqueueApplicationMessage(this, _willMessage.ToPublishPacket()); + _sessionsManager.EnqueueApplicationMessage(this, _willMessage); } _willMessage = null; @@ -211,24 +211,20 @@ namespace MQTTnet.Server } } - public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) + public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - var checkSubscriptionsResult = _subscriptionsManager.CheckSubscriptions(publishPacket.Topic, publishPacket.QualityOfServiceLevel); + var checkSubscriptionsResult = _subscriptionsManager.CheckSubscriptions(applicationMessage.Topic, applicationMessage.QualityOfServiceLevel); if (!checkSubscriptionsResult.IsSubscribed) { return; } - publishPacket = new MqttPublishPacket - { - Topic = publishPacket.Topic, - Payload = publishPacket.Payload, - QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel, - Retain = false, - Dup = false - }; + var publishPacket = _adapter.PacketFormatterAdapter.ConvertApplicationMessageToPublishPacket(applicationMessage); + publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel; + publishPacket.Retain = false; + publishPacket.Dup = false; if (publishPacket.QualityOfServiceLevel > 0) { @@ -370,7 +366,7 @@ namespace MQTTnet.Server var retainedMessages = _retainedMessagesManager.GetSubscribedMessages(topicFilters); foreach (var applicationMessage in retainedMessages) { - EnqueueApplicationMessage(null, applicationMessage.ToPublishPacket()); + EnqueueApplicationMessage(null, applicationMessage); } } @@ -422,7 +418,7 @@ namespace MQTTnet.Server private void HandleIncomingPublishPacketWithQoS0(MqttPublishPacket publishPacket) { - _sessionsManager.EnqueueApplicationMessage(this, publishPacket); + _sessionsManager.EnqueueApplicationMessage(this, publishPacket.ToApplicationMessage()); } private void HandleIncomingPublishPacketWithQoS1( @@ -430,7 +426,7 @@ namespace MQTTnet.Server MqttPublishPacket publishPacket, CancellationToken cancellationToken) { - _sessionsManager.EnqueueApplicationMessage(this, publishPacket); + _sessionsManager.EnqueueApplicationMessage(this, publishPacket.ToApplicationMessage()); var response = new MqttPubAckPacket { @@ -446,7 +442,7 @@ namespace MQTTnet.Server CancellationToken cancellationToken) { // QoS 2 is implement as method "B" (4.3.3 QoS 2: Exactly once delivery) - _sessionsManager.EnqueueApplicationMessage(this, publishPacket); + _sessionsManager.EnqueueApplicationMessage(this, publishPacket.ToApplicationMessage()); var response = new MqttPubRecPacket { diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 7c55109f..0b886363 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -80,11 +80,11 @@ namespace MQTTnet.Server return result; } - public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) + public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) { - if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - _messageQueue.Add(new MqttEnqueuedApplicationMessage(senderClientSession, publishPacket), _cancellationToken); + _messageQueue.Add(new MqttEnqueuedApplicationMessage(senderClientSession, applicationMessage), _cancellationToken); } public Task SubscribeAsync(string clientId, IList topicFilters) @@ -157,8 +157,9 @@ namespace MQTTnet.Server try { var enqueuedApplicationMessage = _messageQueue.Take(cancellationToken); + var sender = enqueuedApplicationMessage.Sender; - var applicationMessage = enqueuedApplicationMessage.PublishPacket.ToApplicationMessage(); + var applicationMessage = enqueuedApplicationMessage.ApplicationMessage; var interceptorContext = InterceptApplicationMessage(sender, applicationMessage); if (interceptorContext != null) @@ -185,7 +186,7 @@ namespace MQTTnet.Server foreach (var clientSession in GetSessions()) { - clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage.ToPublishPacket()); + clientSession.EnqueueApplicationMessage(enqueuedApplicationMessage.Sender, applicationMessage); } } catch (OperationCanceledException) diff --git a/Source/MQTTnet/Server/MqttEnqueuedApplicationMessage.cs b/Source/MQTTnet/Server/MqttEnqueuedApplicationMessage.cs index 37591d89..20ff2feb 100644 --- a/Source/MQTTnet/Server/MqttEnqueuedApplicationMessage.cs +++ b/Source/MQTTnet/Server/MqttEnqueuedApplicationMessage.cs @@ -1,17 +1,15 @@ -using MQTTnet.Packets; - -namespace MQTTnet.Server +namespace MQTTnet.Server { public class MqttEnqueuedApplicationMessage { - public MqttEnqueuedApplicationMessage(MqttClientSession sender, MqttPublishPacket publishPacket) + public MqttEnqueuedApplicationMessage(MqttClientSession sender, MqttApplicationMessage applicationMessage) { Sender = sender; - PublishPacket = publishPacket; + ApplicationMessage = applicationMessage; } public MqttClientSession Sender { get; } - public MqttPublishPacket PublishPacket { get; } + public MqttApplicationMessage ApplicationMessage { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index a960ab00..e0f5a53d 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -76,7 +76,7 @@ namespace MQTTnet.Server if (_cancellationTokenSource == null) throw new InvalidOperationException("The server is not started."); - _clientSessionsManager.EnqueueApplicationMessage(null, applicationMessage.ToPublishPacket()); + _clientSessionsManager.EnqueueApplicationMessage(null, applicationMessage); return Task.FromResult(0); } diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs index 4461a135..9edcdc2e 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs @@ -17,7 +17,7 @@ namespace MQTTnet.AspNetCore.Tests [TestMethod] public async Task TestReceivePacketAsyncThrowsWhenReaderCompleted() { - var serializer = new MqttPacketSerializerAdapter(MqttProtocolVersion.V311); + var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311); var pipe = new DuplexPipeMockup(); var connection = new DefaultConnectionContext(); connection.Transport = pipe; @@ -31,7 +31,7 @@ namespace MQTTnet.AspNetCore.Tests [TestMethod] public async Task TestParallelWrites() { - var serializer = new MqttPacketSerializerAdapter(MqttProtocolVersion.V311); + var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311); var pipe = new DuplexPipeMockup(); var connection = new DefaultConnectionContext(); connection.Transport = pipe; diff --git a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs index 7d5e55ac..1db41c78 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs @@ -2,6 +2,7 @@ using System.Buffers; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.AspNetCore; +using MQTTnet.Formatter; using MQTTnet.Formatter.V311; using MQTTnet.Packets; @@ -13,7 +14,7 @@ namespace MQTTnet.AspNetCore.Tests [TestMethod] public void TestTryDeserialize() { - var serializer = new MqttV311PacketFormatter(); + var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311); var buffer = serializer.Encode(new MqttPublishPacket() {Topic = "a", Payload = new byte[5]}); @@ -26,20 +27,20 @@ namespace MQTTnet.AspNetCore.Tests var result = false; part = sequence.Slice(sequence.Start, 0); // empty message should fail - result = serializer.TryDeserialize(part, out packet, out consumed, out observed); + result = serializer.TryDecode(part, out packet, out consumed, out observed); Assert.IsFalse(result); part = sequence.Slice(sequence.Start, 1); // partial fixed header should fail - result = serializer.TryDeserialize(part, out packet, out consumed, out observed); + result = serializer.TryDecode(part, out packet, out consumed, out observed); Assert.IsFalse(result); part = sequence.Slice(sequence.Start, 4); // partial body should fail - result = serializer.TryDeserialize(part, out packet, out consumed, out observed); + result = serializer.TryDecode(part, out packet, out consumed, out observed); Assert.IsFalse(result); part = sequence; // complete msg should work - result = serializer.TryDeserialize(part, out packet, out consumed, out observed); + result = serializer.TryDecode(part, out packet, out consumed, out observed); Assert.IsTrue(result); } } diff --git a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs index fb7443cb..ea035700 100644 --- a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs @@ -21,12 +21,12 @@ namespace MQTTnet.Benchmarks [GlobalSetup] public void Setup() { - var message = new MqttApplicationMessageBuilder() - .WithTopic("A") - .Build(); + _packet = new MqttPublishPacket + { + Topic = "A" + }; - _packet = message.ToPublishPacket(); - var serializer = new MqttPacketSerializerAdapter(MqttProtocolVersion.V311); + var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311); var serializedPacket = Join(serializer.Encode(_packet)); diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj index 33f08667..26e2415f 100644 --- a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -8,9 +8,9 @@ - + - + diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 53caec15..c7fdb99d 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -1,6 +1,5 @@ using BenchmarkDotNet.Attributes; using MQTTnet.Packets; -using MQTTnet.Internal; using System; using System.Threading; using System.Threading.Tasks; @@ -23,11 +22,11 @@ namespace MQTTnet.Benchmarks [GlobalSetup] public void Setup() { - var message = new MqttApplicationMessageBuilder() - .WithTopic("A") - .Build(); + _packet = new MqttPublishPacket + { + Topic = "A" + }; - _packet = message.ToPublishPacket(); _serializer = new MqttV311PacketFormatter(); _serializedPacket = _serializer.Encode(_packet); } @@ -47,17 +46,17 @@ namespace MQTTnet.Benchmarks { var channel = new BenchmarkMqttChannel(_serializedPacket); var fixedHeader = new byte[2]; - var singleByteBuffer = new byte[1]; + var reader = new MqttPacketReader(channel); for (var i = 0; i < 10000; i++) { channel.Reset(); - var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); + var header = reader.ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult(); var receivedPacket = new ReceivedMqttPacket( header.Flags, - new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength, _serializedPacket.Array.Length)); + new MqttPacketBodyReader(_serializedPacket.Array, (ulong)(_serializedPacket.Count - header.RemainingLength), (ulong)_serializedPacket.Array.Length)); _serializer.Decode(receivedPacket); } diff --git a/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs b/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs index a36555a4..183955a1 100644 --- a/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs @@ -67,7 +67,7 @@ namespace MQTTnet.Core.Tests throw new NotSupportedException(); } - public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket) + public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) { throw new NotSupportedException(); } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs index eb2ad8f7..51c5e073 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs @@ -15,8 +15,8 @@ namespace MQTTnet.Core.Tests public void MqttPacketReader_EmptyStream() { var fixedHeader = new byte[2]; - var singleByteBuffer = new byte[1]; - MqttPacketReader.ReadFixedHeaderAsync(new TestMqttChannel(new MemoryStream()), fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); + var reader = new MqttPacketReader(new TestMqttChannel(new MemoryStream())); + reader.ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult(); } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index ebb379f6..f8bffcdf 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -181,17 +181,15 @@ namespace MQTTnet.Core.Tests var buffer = serializer.Encode(publishPacket); var testChannel = new TestMqttChannel(new MemoryStream(buffer.Array, buffer.Offset, buffer.Count)); - var header = MqttPacketReader.ReadFixedHeaderAsync( - testChannel, + var header = new MqttPacketReader(testChannel).ReadFixedHeaderAsync( new byte[2], - new byte[1], CancellationToken.None).GetAwaiter().GetResult(); var eof = buffer.Offset + buffer.Count; var receivedPacket = new ReceivedMqttPacket( header.Flags, - new MqttPacketBodyReader(buffer.Array, eof - header.RemainingLength, buffer.Count + buffer.Offset)); + new MqttPacketBodyReader(buffer.Array, (uint)eof - header.RemainingLength, (uint)(buffer.Count + buffer.Offset))); var packet = (MqttPublishPacket)serializer.Decode(receivedPacket); @@ -552,10 +550,9 @@ namespace MQTTnet.Core.Tests { var channel = new TestMqttChannel(headerStream); var fixedHeader = new byte[2]; - var singleByteBuffer = new byte[1]; - var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); + var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult(); - using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) + using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, (int)header.RemainingLength)) { var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length))); var buffer2 = serializer.Encode(deserializedPacket); @@ -588,11 +585,10 @@ namespace MQTTnet.Core.Tests { var channel = new TestMqttChannel(headerStream); var fixedHeader = new byte[2]; - var singleByteBuffer = new byte[1]; - var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); + var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult(); - using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) + 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))); } diff --git a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs index ceb44e13..fd95cfae 100644 --- a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs @@ -16,7 +16,7 @@ namespace MQTTnet.Core.Tests public string Endpoint { get; } - public MqttPacketSerializerAdapter PacketSerializerAdapter { get; } = new MqttPacketSerializerAdapter(MqttProtocolVersion.V311); + public MqttPacketFormatterAdapter PacketFormatterAdapter { get; } = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311); public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj index a23f309b..b8a9307f 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj +++ b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj @@ -10,8 +10,8 @@ - - + + diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj index f8583aa1..551bd2eb 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj @@ -142,10 +142,10 @@ - 6.1.9 + 6.2.2 - 4.0.0 + 5.0.0