diff --git a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs index e308ccd4..93253857 100644 --- a/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs +++ b/Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs @@ -14,15 +14,13 @@ namespace MQTTnet.AspNetCore.Client { if (options == null) throw new ArgumentNullException(nameof(options)); - var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }; - switch (options.ChannelOptions) { case MqttClientTcpOptions tcpOptions: { var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort()); var tcpConnection = new TcpConnection(endpoint); - return new MqttConnectionContext(serializer, tcpConnection); + return new MqttConnectionContext(new MqttPacketSerializerAdapter(options.ProtocolVersion), tcpConnection); } default: { diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 8a8469de..2d27894d 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -12,17 +12,15 @@ namespace MQTTnet.AspNetCore { public class MqttConnectionContext : IMqttChannelAdapter { - public MqttConnectionContext( - IMqttPacketSerializer packetSerializer, - ConnectionContext connection) + public MqttConnectionContext(MqttPacketSerializerAdapter packetSerializerAdapter, ConnectionContext connection) { - PacketSerializer = packetSerializer; - Connection = connection; + PacketSerializerAdapter = packetSerializerAdapter ?? throw new ArgumentNullException(nameof(packetSerializerAdapter)); + Connection = connection ?? throw new ArgumentNullException(nameof(connection)); } public string Endpoint => Connection.ConnectionId; public ConnectionContext Connection { get; } - public IMqttPacketSerializer PacketSerializer { get; } + public MqttPacketSerializerAdapter PacketSerializerAdapter { get; } public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; @@ -71,7 +69,7 @@ namespace MQTTnet.AspNetCore { if (!buffer.IsEmpty) { - if (PacketSerializer.TryDeserialize(buffer, out var packet, out consumed, out observed)) + if (PacketSerializerAdapter.Serializer.TryDeserialize(buffer, out var packet, out consumed, out observed)) { return packet; } @@ -106,7 +104,7 @@ namespace MQTTnet.AspNetCore public Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) { - var buffer = PacketSerializer.Serialize(packet); + var buffer = PacketSerializerAdapter.Serialize(packet); return Connection.Transport.Output.WriteAsync(buffer.AsMemory(), cancellationToken).AsTask(); } diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs index 23903894..77a8734b 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs @@ -9,7 +9,7 @@ using System.Threading.Tasks; namespace MQTTnet.AspNetCore { public class MqttConnectionHandler : ConnectionHandler, IMqttServerAdapter - { + { public event EventHandler ClientAccepted; public override async Task OnConnectedAsync(ConnectionContext connection) @@ -21,9 +21,7 @@ namespace MQTTnet.AspNetCore transferFormatFeature.ActiveFormat = TransferFormat.Binary; } - - var serializer = new MqttPacketSerializer(); - using (var adapter = new MqttConnectionContext(serializer, connection)) + using (var adapter = new MqttConnectionContext(new MqttPacketSerializerAdapter(), connection)) { var args = new MqttServerAdapterClientAcceptedEventArgs(adapter); ClientAccepted?.Invoke(this, args); @@ -41,7 +39,7 @@ namespace MQTTnet.AspNetCore { return Task.CompletedTask; } - + public void Dispose() { } diff --git a/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs b/Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs index dbd93604..4dc75a44 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 MqttPacketSerializer(), new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); + var clientAdapter = new MqttChannelAdapter(new MqttWebSocketChannel(webSocket, endpoint), new MqttPacketSerializerAdapter(), new MqttNetLogger().CreateChildLogger(nameof(MqttWebSocketServerAdapter))); var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter); ClientAccepted?.Invoke(this, eventArgs); @@ -37,7 +37,7 @@ namespace MQTTnet.AspNetCore await eventArgs.SessionTask.ConfigureAwait(false); } } - + public void Dispose() { StopAsync().GetAwaiter().GetResult(); diff --git a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs index 43e5c8c4..9aa5cacc 100644 --- a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Adapter { string Endpoint { get; } - IMqttPacketSerializer PacketSerializer { get; } + MqttPacketSerializerAdapter PacketSerializerAdapter { get; } event EventHandler ReadingPacketStarted; diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 4e617330..01d252b0 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -22,26 +22,25 @@ namespace MQTTnet.Adapter private readonly IMqttNetChildLogger _logger; private readonly IMqttChannel _channel; - + private readonly byte[] _fixedHeaderBuffer = new byte[2]; - private readonly byte[] _singleByteBuffer = new byte[1]; private bool _isDisposed; - public MqttChannelAdapter(IMqttChannel channel, IMqttPacketSerializer serializer, IMqttNetChildLogger logger) + public MqttChannelAdapter(IMqttChannel channel, MqttPacketSerializerAdapter packetSerializerAdapter, IMqttNetChildLogger logger) { if (logger == null) throw new ArgumentNullException(nameof(logger)); _channel = channel ?? throw new ArgumentNullException(nameof(channel)); - PacketSerializer = serializer ?? throw new ArgumentNullException(nameof(serializer)); - + PacketSerializerAdapter = packetSerializerAdapter ?? throw new ArgumentNullException(nameof(packetSerializerAdapter)); + _logger = logger.CreateChildLogger(nameof(MqttChannelAdapter)); } public string Endpoint => _channel.Endpoint; - public IMqttPacketSerializer PacketSerializer { get; } + public MqttPacketSerializerAdapter PacketSerializerAdapter { get; } public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; @@ -91,15 +90,15 @@ namespace MQTTnet.Adapter WrapException(exception); } } - + public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) { await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); try { - var packetData = PacketSerializer.Serialize(packet); + var packetData = PacketSerializerAdapter.Serialize(packet); await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false); - PacketSerializer.FreeBuffer(); + PacketSerializerAdapter.FreeBuffer(); _logger.Verbose("TX >>> {0}", packet); } @@ -140,7 +139,12 @@ namespace MQTTnet.Adapter return null; } - var packet = PacketSerializer.Deserialize(receivedMqttPacket); + if (!PacketSerializerAdapter.ProtocolVersion.HasValue) + { + PacketSerializerAdapter.DetectProtocolVersion(receivedMqttPacket); + } + + var packet = PacketSerializerAdapter.Deserialize(receivedMqttPacket); if (packet == null) { throw new MqttProtocolViolationException("Received malformed packet."); diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 0fa074c2..643d0ef9 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -9,6 +9,7 @@ using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Internal; using MQTTnet.Packets; +using MQTTnet.Packets.V3; using MQTTnet.Protocol; namespace MQTTnet.Client @@ -61,7 +62,7 @@ namespace MQTTnet.Client _packetDispatcher.Reset(); _adapter = _adapterFactory.CreateClientAdapter(options, _logger); - + _logger.Verbose($"Trying to connect with server ({_options.ChannelOptions})."); await _adapter.ConnectAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token).ConfigureAwait(false); _logger.Verbose("Connection with server established."); @@ -189,9 +190,9 @@ namespace MQTTnet.Client _adapter?.Dispose(); } - private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) + private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) { - var connectPacket = new MqttConnectPacket + var connectPacket = new MqttV3ConnectPacket { ClientId = _options.ClientId, Username = _options.Credentials?.Username, @@ -201,7 +202,7 @@ namespace MQTTnet.Client WillMessage = willApplicationMessage }; - var response = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false); + var response = await SendAndReceiveAsync(connectPacket, cancellationToken).ConfigureAwait(false); if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { throw new MqttConnectingFailedException(response.ConnectReturnCode); diff --git a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs index dd06c31a..169d18e2 100644 --- a/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs +++ b/Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs @@ -11,19 +11,17 @@ namespace MQTTnet.Implementations public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger) { if (options == null) throw new ArgumentNullException(nameof(options)); - - var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }; - + switch (options.ChannelOptions) { case MqttClientTcpOptions _: { - return new MqttChannelAdapter(new MqttTcpChannel(options), serializer, logger); + return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketSerializerAdapter(options.ProtocolVersion), logger); } case MqttClientWebSocketOptions webSocketOptions: { - return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), serializer, logger); + return new MqttChannelAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializerAdapter(options.ProtocolVersion), logger); } default: diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs index c42db2b6..d494b910 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 MqttPacketSerializer(), _logger); + var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, _options), new MqttPacketSerializerAdapter(), _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 e3522979..d91b66e6 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -83,7 +83,7 @@ namespace MQTTnet.Implementations _socket.LocalEndPoint, _addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6"); - var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger); + var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializerAdapter(), _logger); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); } catch (ObjectDisposedException) diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 30eace10..1704d7d0 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -63,4 +63,8 @@ + + + + \ No newline at end of file diff --git a/Source/MQTTnet/Packets/MqttConnAckPacket.cs b/Source/MQTTnet/Packets/MqttConnAckPacket.cs index dbdeea89..a0f69fdd 100644 --- a/Source/MQTTnet/Packets/MqttConnAckPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnAckPacket.cs @@ -1,16 +1,6 @@ -using MQTTnet.Protocol; - -namespace MQTTnet.Packets +namespace MQTTnet.Packets { - public class MqttConnAckPacket : MqttBasePacket + public abstract class MqttConnAckPacket : MqttBasePacket { - public bool IsSessionPresent { get; set; } - - public MqttConnectReturnCode ConnectReturnCode { get; set; } - - public override string ToString() - { - return "ConnAck: [ConnectReturnCode=" + ConnectReturnCode + "] [IsSessionPresent=" + IsSessionPresent + "]"; - } } } diff --git a/Source/MQTTnet/Packets/MqttConnectPacket.cs b/Source/MQTTnet/Packets/MqttConnectPacket.cs index 2134f4bb..1faa8c20 100644 --- a/Source/MQTTnet/Packets/MqttConnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnectPacket.cs @@ -1,26 +1,6 @@ -using MQTTnet.Serializer; - -namespace MQTTnet.Packets +namespace MQTTnet.Packets { - public class MqttConnectPacket : MqttBasePacket + public abstract class MqttConnectPacket : MqttBasePacket { - public MqttProtocolVersion ProtocolVersion { get; set; } - - public string ClientId { get; set; } - - public string Username { get; set; } - - public string Password { get; set; } - - public ushort KeepAlivePeriod { get; set; } - - public bool CleanSession { get; set; } - - public MqttApplicationMessage WillMessage { get; set; } - - public override string ToString() - { - return "Connect: [ClientId=" + ClientId + "] [Username=" + Username + "] [Password=" + Password + "] [KeepAlivePeriod=" + KeepAlivePeriod + "] [CleanSession=" + CleanSession + "]"; - } } } diff --git a/Source/MQTTnet/Packets/V3/MqttV3ConnAckPacket.cs b/Source/MQTTnet/Packets/V3/MqttV3ConnAckPacket.cs new file mode 100644 index 00000000..3e858ec7 --- /dev/null +++ b/Source/MQTTnet/Packets/V3/MqttV3ConnAckPacket.cs @@ -0,0 +1,16 @@ +using MQTTnet.Protocol; + +namespace MQTTnet.Packets.V3 +{ + public class MqttV3ConnAckPacket : MqttBasePacket + { + public bool IsSessionPresent { get; set; } + + public MqttConnectReturnCode ConnectReturnCode { get; set; } + + public override string ToString() + { + return "ConnAck: [ConnectReturnCode=" + ConnectReturnCode + "] [IsSessionPresent=" + IsSessionPresent + "]"; + } + } +} diff --git a/Source/MQTTnet/Packets/V3/MqttV3ConnectPacket.cs b/Source/MQTTnet/Packets/V3/MqttV3ConnectPacket.cs new file mode 100644 index 00000000..2434d594 --- /dev/null +++ b/Source/MQTTnet/Packets/V3/MqttV3ConnectPacket.cs @@ -0,0 +1,26 @@ +namespace MQTTnet.Packets.V3 +{ + public class MqttV3ConnectPacket : MqttConnectPacket + { + public string ProtocolName { get; set; } + + public byte ProtocolLevel { get; set; } + + public string ClientId { get; set; } + + public string Username { get; set; } + + public string Password { get; set; } + + public ushort KeepAlivePeriod { get; set; } + + public bool CleanSession { get; set; } + + public MqttApplicationMessage WillMessage { get; set; } + + public override string ToString() + { + return "Connect: [ProtocolLevel=" + ProtocolLevel + "] [ClientId=" + ClientId + "] [Username=" + Username + "] [Password=" + Password + "] [KeepAlivePeriod=" + KeepAlivePeriod + "] [CleanSession=" + CleanSession + "]"; + } + } +} diff --git a/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs b/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs index e3066f9f..47d1cd8e 100644 --- a/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs +++ b/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs @@ -6,8 +6,6 @@ namespace MQTTnet.Serializer { public interface IMqttPacketSerializer { - MqttProtocolVersion ProtocolVersion { get; set; } - ArraySegment Serialize(MqttBasePacket mqttPacket); MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket); diff --git a/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs b/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs index f0e1b562..ad3d789b 100644 --- a/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs +++ b/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs @@ -6,6 +6,7 @@ namespace MQTTnet.Serializer public class MqttPacketBodyReader { private readonly byte[] _buffer; + private readonly int _initialOffset; private readonly int _length; private int _offset; @@ -13,6 +14,7 @@ namespace MQTTnet.Serializer public MqttPacketBodyReader(byte[] buffer, int offset, int length) { _buffer = buffer; + _initialOffset = offset; _offset = offset; _length = length; } @@ -21,6 +23,11 @@ namespace MQTTnet.Serializer public bool EndOfStream => _offset == _length; + public void Seek(int position) + { + _offset = _initialOffset + position; + } + public byte ReadByte() { ValidateReceiveBuffer(1); diff --git a/Source/MQTTnet/Serializer/MqttPacketSerializerAdapter.cs b/Source/MQTTnet/Serializer/MqttPacketSerializerAdapter.cs new file mode 100644 index 00000000..9a681e90 --- /dev/null +++ b/Source/MQTTnet/Serializer/MqttPacketSerializerAdapter.cs @@ -0,0 +1,123 @@ +using System; +using MQTTnet.Adapter; +using MQTTnet.Exceptions; +using MQTTnet.Packets; + +namespace MQTTnet.Serializer +{ + public class MqttPacketSerializerAdapter + { + public MqttPacketSerializerAdapter() + { + } + + public MqttPacketSerializerAdapter(MqttProtocolVersion protocolVersion) + { + UseProtocolVersion(protocolVersion); + } + + public MqttProtocolVersion? ProtocolVersion { get; private set; } + + public IMqttPacketSerializer Serializer { get; private set; } + + public ArraySegment Serialize(MqttBasePacket packet) + { + if (Serializer == null) + { + throw new InvalidOperationException("Protocol version not set or detected."); + } + + return Serializer.Serialize(packet); + } + + public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket) + { + if (Serializer == null) + { + throw new InvalidOperationException("Protocol version nor set or detected."); + } + + return Serializer.Deserialize(receivedMqttPacket); + } + + public void FreeBuffer() + { + Serializer?.FreeBuffer(); + } + + public void DetectProtocolVersion(ReceivedMqttPacket receivedMqttPacket) + { + var protocolVersion = ParseProtocolVersion(receivedMqttPacket); + receivedMqttPacket.Body.Seek(0); + + UseProtocolVersion(protocolVersion); + } + + private void UseProtocolVersion(MqttProtocolVersion protocolVersion) + { + ProtocolVersion = protocolVersion; + + switch (protocolVersion) + { + case MqttProtocolVersion.V500: + { + Serializer = new MqttV500PacketSerializer(); + break; + } + + case MqttProtocolVersion.V311: + { + Serializer = new MqttV311PacketSerializer(); + break; + } + + case MqttProtocolVersion.V310: + { + + Serializer = new MqttV310PacketSerializer(); + break; + } + + default: + { + throw new NotSupportedException(); + } + } + } + + private MqttProtocolVersion ParseProtocolVersion(ReceivedMqttPacket receivedMqttPacket) + { + if (receivedMqttPacket == null) throw new ArgumentNullException(nameof(receivedMqttPacket)); + + var protocolName = receivedMqttPacket.Body.ReadStringWithLengthPrefix(); + var protocolLevel = receivedMqttPacket.Body.ReadByte(); + + if (protocolName == "MQTT") + { + if (protocolLevel == 5) + { + return MqttProtocolVersion.V500; + } + + if (protocolLevel == 4) + { + return MqttProtocolVersion.V311; + } + + throw new MqttProtocolViolationException($"Protocol level '{protocolLevel}' not supported."); + } + + if (protocolName == "MQIsdp") + { + if (protocolLevel == 3) + { + return MqttProtocolVersion.V310; + } + + throw new MqttProtocolViolationException($"Protocol level '{protocolLevel}' not supported."); + } + + throw new MqttProtocolViolationException($"Protocol '{protocolName}' not supported."); + } + } +} diff --git a/Source/MQTTnet/Serializer/MqttPacketWriter.cs b/Source/MQTTnet/Serializer/MqttPacketWriter.cs index 67fd4672..10ec4f43 100644 --- a/Source/MQTTnet/Serializer/MqttPacketWriter.cs +++ b/Source/MQTTnet/Serializer/MqttPacketWriter.cs @@ -100,10 +100,10 @@ namespace MQTTnet.Serializer Length = 5; } - public void Seek(int offset) + public void Seek(int position) { - EnsureCapacity(offset); - _position = offset; + EnsureCapacity(position); + _position = position; } public byte[] GetBuffer() diff --git a/Source/MQTTnet/Serializer/MqttProtocolVersion.cs b/Source/MQTTnet/Serializer/MqttProtocolVersion.cs index 15e698d3..576f7df7 100644 --- a/Source/MQTTnet/Serializer/MqttProtocolVersion.cs +++ b/Source/MQTTnet/Serializer/MqttProtocolVersion.cs @@ -2,7 +2,8 @@ { public enum MqttProtocolVersion { - V311, - V310 + V310 = 3, + V311 = 4, + V500 = 5 } } diff --git a/Source/MQTTnet/Serializer/MqttPacketSerializer.cs b/Source/MQTTnet/Serializer/MqttV310PacketSerializer.cs similarity index 79% rename from Source/MQTTnet/Serializer/MqttPacketSerializer.cs rename to Source/MQTTnet/Serializer/MqttV310PacketSerializer.cs index f1eb790a..572feb9b 100644 --- a/Source/MQTTnet/Serializer/MqttPacketSerializer.cs +++ b/Source/MQTTnet/Serializer/MqttV310PacketSerializer.cs @@ -4,17 +4,16 @@ using MQTTnet.Protocol; using System; using System.Linq; using MQTTnet.Adapter; +using MQTTnet.Packets.V3; namespace MQTTnet.Serializer { - public class MqttPacketSerializer : IMqttPacketSerializer + public class MqttV310PacketSerializer : IMqttPacketSerializer { private const int FixedHeaderSize = 1; private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter(); - public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; - public ArraySegment Serialize(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -52,8 +51,8 @@ namespace MQTTnet.Serializer switch ((MqttControlPacketType)controlPacketType) { - case MqttControlPacketType.Connect: return DeserializeConnect(receivedMqttPacket.Body); - case MqttControlPacketType.ConnAck: return DeserializeConnAck(receivedMqttPacket.Body); + case MqttControlPacketType.Connect: return DeserializeConnectPacket(receivedMqttPacket.Body); + case MqttControlPacketType.ConnAck: return DeserializeConnAckPacket(receivedMqttPacket.Body); case MqttControlPacketType.Disconnect: return new MqttDisconnectPacket(); case MqttControlPacketType.Publish: return DeserializePublish(receivedMqttPacket); case MqttControlPacketType.PubAck: return DeserializePubAck(receivedMqttPacket.Body); @@ -63,7 +62,7 @@ namespace MQTTnet.Serializer case MqttControlPacketType.PingReq: return new MqttPingReqPacket(); case MqttControlPacketType.PingResp: return new MqttPingRespPacket(); case MqttControlPacketType.Subscribe: return DeserializeSubscribe(receivedMqttPacket.Body); - case MqttControlPacketType.SubAck: return DeserializeSubAck(receivedMqttPacket.Body); + case MqttControlPacketType.SubAck: return DeserializeSubAckPacket(receivedMqttPacket.Body); case MqttControlPacketType.Unsubscibe: return DeserializeUnsubscribe(receivedMqttPacket.Body); case MqttControlPacketType.UnsubAck: return DeserializeUnsubAck(receivedMqttPacket.Body); @@ -80,20 +79,20 @@ namespace MQTTnet.Serializer { switch (packet) { - case MqttConnectPacket connectPacket: return Serialize(connectPacket, packetWriter); - case MqttConnAckPacket connAckPacket: return Serialize(connAckPacket, packetWriter); + case MqttV3ConnectPacket connectPacket: return SerializeConnectPacket(connectPacket, packetWriter); + case MqttV3ConnAckPacket connAckPacket: return SerializeConnAckPacket(connAckPacket, packetWriter); case MqttDisconnectPacket _: return SerializeEmptyPacket(MqttControlPacketType.Disconnect); case MqttPingReqPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingReq); case MqttPingRespPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingResp); - case MqttPublishPacket publishPacket: return Serialize(publishPacket, packetWriter); - case MqttPubAckPacket pubAckPacket: return Serialize(pubAckPacket, packetWriter); - case MqttPubRecPacket pubRecPacket: return Serialize(pubRecPacket, packetWriter); - case MqttPubRelPacket pubRelPacket: return Serialize(pubRelPacket, packetWriter); - case MqttPubCompPacket pubCompPacket: return Serialize(pubCompPacket, packetWriter); - case MqttSubscribePacket subscribePacket: return Serialize(subscribePacket, packetWriter); - case MqttSubAckPacket subAckPacket: return Serialize(subAckPacket, packetWriter); - case MqttUnsubscribePacket unsubscribePacket: return Serialize(unsubscribePacket, packetWriter); - case MqttUnsubAckPacket unsubAckPacket: return Serialize(unsubAckPacket, packetWriter); + case MqttPublishPacket publishPacket: return SerializePublishPacket(publishPacket, packetWriter); + case MqttPubAckPacket pubAckPacket: return SerializePubAckPacket(pubAckPacket, packetWriter); + case MqttPubRecPacket pubRecPacket: return SerializePubRecPacket(pubRecPacket, packetWriter); + case MqttPubRelPacket pubRelPacket: return SerializePubRelPacket(pubRelPacket, packetWriter); + case MqttPubCompPacket pubCompPacket: return SerializePubCompPacket(pubCompPacket, packetWriter); + case MqttSubscribePacket subscribePacket: return SerializeSubscribePacket(subscribePacket, packetWriter); + case MqttSubAckPacket subAckPacket: return SerializeSubAckPacket(subAckPacket, packetWriter); + case MqttUnsubscribePacket unsubscribePacket: return SerializeUnsubscribePacket(unsubscribePacket, packetWriter); + case MqttUnsubAckPacket unsubAckPacket: return SerializeUnsubAckPacket(unsubAckPacket, packetWriter); default: throw new MqttProtocolViolationException("Packet type invalid."); } } @@ -213,49 +212,23 @@ namespace MQTTnet.Serializer return packet; } - private static MqttBasePacket DeserializeConnect(MqttPacketBodyReader body) + private MqttBasePacket DeserializeConnectPacket(MqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); - var protocolName = body.ReadStringWithLengthPrefix(); - - MqttProtocolVersion protocolVersion; - if (protocolName == "MQTT") + var packet = new MqttV3ConnectPacket { - var protocolLevel = body.ReadByte(); - if (protocolLevel != 4) - { - throw new MqttProtocolViolationException($"Protocol level ({protocolLevel}) not supported for MQTT 3.1.1."); - } - - protocolVersion = MqttProtocolVersion.V311; - } - else if (protocolName == "MQIsdp") - { - var protocolLevel = body.ReadByte(); - if (protocolLevel != 3) - { - throw new MqttProtocolViolationException($"Protocol level ({protocolLevel}) not supported for MQTT 3.1."); - } - - protocolVersion = MqttProtocolVersion.V310; - } - else - { - throw new MqttProtocolViolationException($"Protocol name ({protocolName}) is not supported."); - } - + ProtocolName = body.ReadStringWithLengthPrefix(), + ProtocolLevel = body.ReadByte() + }; + var connectFlags = body.ReadByte(); if ((connectFlags & 0x1) > 0) { throw new MqttProtocolViolationException("The first bit of the Connect Flags must be set to 0."); } - var packet = new MqttConnectPacket - { - ProtocolVersion = protocolVersion, - CleanSession = (connectFlags & 0x2) > 0 - }; + packet.CleanSession = (connectFlags & 0x2) > 0; var willFlag = (connectFlags & 0x4) > 0; var willQoS = (connectFlags & 0x18) >> 3; @@ -291,7 +264,7 @@ namespace MQTTnet.Serializer return packet; } - private static MqttBasePacket DeserializeSubAck(MqttPacketBodyReader body) + private static MqttBasePacket DeserializeSubAckPacket(MqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -308,25 +281,19 @@ namespace MQTTnet.Serializer return packet; } - private MqttBasePacket DeserializeConnAck(MqttPacketBodyReader body) + protected virtual MqttBasePacket DeserializeConnAckPacket(MqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); - var packet = new MqttConnAckPacket(); - - var acknowledgeFlags = body.ReadByte(); - - if (ProtocolVersion == MqttProtocolVersion.V311) - { - packet.IsSessionPresent = (acknowledgeFlags & 0x1) > 0; - } + var packet = new MqttV3ConnAckPacket(); + body.ReadByte(); // Reserved. packet.ConnectReturnCode = (MqttConnectReturnCode)body.ReadByte(); return packet; } - private static void ValidateConnectPacket(MqttConnectPacket packet) + protected void ValidateConnectPacket(MqttV3ConnectPacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -345,21 +312,12 @@ namespace MQTTnet.Serializer } } - private byte Serialize(MqttConnectPacket packet, MqttPacketWriter packetWriter) + protected virtual byte SerializeConnectPacket(MqttV3ConnectPacket packet, MqttPacketWriter packetWriter) { ValidateConnectPacket(packet); - // Write variable header - if (ProtocolVersion == MqttProtocolVersion.V311) - { - packetWriter.WriteWithLengthPrefix("MQTT"); - packetWriter.Write(4); // 3.1.2.2 Protocol Level 4 - } - else - { - packetWriter.WriteWithLengthPrefix("MQIsdp"); - packetWriter.Write(3); // Protocol Level 3 - } + packetWriter.WriteWithLengthPrefix("MQIsdp"); + packetWriter.Write(3); // Protocol Level 3 byte connectFlags = 0x0; if (packet.CleanSession) @@ -416,33 +374,15 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); } - private byte Serialize(MqttConnAckPacket packet, MqttPacketWriter packetWriter) + protected virtual byte SerializeConnAckPacket(MqttV3ConnAckPacket packet, MqttPacketWriter packetWriter) { - if (ProtocolVersion == MqttProtocolVersion.V310) - { - packetWriter.Write(0); - } - else if (ProtocolVersion == MqttProtocolVersion.V311) - { - byte connectAcknowledgeFlags = 0x0; - if (packet.IsSessionPresent) - { - connectAcknowledgeFlags |= 0x1; - } - - packetWriter.Write(connectAcknowledgeFlags); - } - else - { - throw new MqttProtocolViolationException("Protocol version not supported."); - } - + packetWriter.Write(0); // Reserved. packetWriter.Write((byte)packet.ConnectReturnCode); return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); } - private static byte Serialize(MqttPubRelPacket packet, MqttPacketWriter packetWriter) + private static byte SerializePubRelPacket(MqttPubRelPacket packet, MqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -454,7 +394,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02); } - private static byte Serialize(MqttPublishPacket packet, MqttPacketWriter packetWriter) + private static byte SerializePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter) { ValidatePublishPacket(packet); @@ -499,7 +439,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader); } - private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter packetWriter) + private static byte SerializePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -511,7 +451,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck); } - private static byte Serialize(MqttPubRecPacket packet, MqttPacketWriter packetWriter) + private static byte SerializePubRecPacket(MqttPubRecPacket packet, MqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -523,7 +463,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec); } - private static byte Serialize(MqttPubCompPacket packet, MqttPacketWriter packetWriter) + private static byte SerializePubCompPacket(MqttPubCompPacket packet, MqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -535,7 +475,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp); } - private static byte Serialize(MqttSubscribePacket packet, MqttPacketWriter packetWriter) + private static byte SerializeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter) { if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3]."); @@ -558,7 +498,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02); } - private static byte Serialize(MqttSubAckPacket packet, MqttPacketWriter packetWriter) + private static byte SerializeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -578,7 +518,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck); } - private static byte Serialize(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter) + private static byte SerializeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter) { if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); @@ -600,7 +540,7 @@ namespace MQTTnet.Serializer return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02); } - private static byte Serialize(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter) + private static byte SerializeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter) { if (!packet.PacketIdentifier.HasValue) { @@ -617,7 +557,7 @@ namespace MQTTnet.Serializer } // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local - private static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body) + protected static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body) { if (body == null || body.Length == 0) { diff --git a/Source/MQTTnet/Serializer/MqttV311PacketSerializer.cs b/Source/MQTTnet/Serializer/MqttV311PacketSerializer.cs new file mode 100644 index 00000000..f68ae02f --- /dev/null +++ b/Source/MQTTnet/Serializer/MqttV311PacketSerializer.cs @@ -0,0 +1,100 @@ +using MQTTnet.Exceptions; +using MQTTnet.Packets; +using MQTTnet.Packets.V3; +using MQTTnet.Protocol; + +namespace MQTTnet.Serializer +{ + public class MqttV311PacketSerializer : MqttV310PacketSerializer + { + protected override byte SerializeConnectPacket(MqttV3ConnectPacket packet, MqttPacketWriter packetWriter) + { + ValidateConnectPacket(packet); + + packetWriter.WriteWithLengthPrefix("MQTT"); + packetWriter.Write(4); // 3.1.2.2 Protocol Level 4 + + 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); + } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); + } + + protected override byte SerializeConnAckPacket(MqttV3ConnAckPacket packet, MqttPacketWriter packetWriter) + { + byte connectAcknowledgeFlags = 0x0; + if (packet.IsSessionPresent) + { + connectAcknowledgeFlags |= 0x1; + } + + packetWriter.Write(connectAcknowledgeFlags); + packetWriter.Write((byte)packet.ConnectReturnCode); + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck); + } + + protected override MqttBasePacket DeserializeConnAckPacket(MqttPacketBodyReader body) + { + ThrowIfBodyIsEmpty(body); + + var packet = new MqttV3ConnAckPacket(); + + var acknowledgeFlags = body.ReadByte(); + + packet.IsSessionPresent = (acknowledgeFlags & 0x1) > 0; + packet.ConnectReturnCode = (MqttConnectReturnCode)body.ReadByte(); + + return packet; + } + } +} diff --git a/Source/MQTTnet/Serializer/MqttV500PacketSerializer.cs b/Source/MQTTnet/Serializer/MqttV500PacketSerializer.cs new file mode 100644 index 00000000..a50a9033 --- /dev/null +++ b/Source/MQTTnet/Serializer/MqttV500PacketSerializer.cs @@ -0,0 +1,24 @@ +using System; +using MQTTnet.Adapter; +using MQTTnet.Packets; + +namespace MQTTnet.Serializer +{ + public class MqttV500PacketSerializer : IMqttPacketSerializer + { + public ArraySegment Serialize(MqttBasePacket mqttPacket) + { + throw new NotImplementedException(); + } + + public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket) + { + throw new NotImplementedException(); + } + + public void FreeBuffer() + { + throw new NotImplementedException(); + } + } +} diff --git a/Source/MQTTnet/Server/IMqttClientSession.cs b/Source/MQTTnet/Server/IMqttClientSession.cs index 9fdb0eb9..e0dd214c 100644 --- a/Source/MQTTnet/Server/IMqttClientSession.cs +++ b/Source/MQTTnet/Server/IMqttClientSession.cs @@ -14,7 +14,7 @@ namespace MQTTnet.Server void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket); void ClearPendingApplicationMessages(); - Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter); + Task RunAsync(MqttApplicationMessage willMessage, int keepAliveInterval, IMqttChannelAdapter adapter); void Stop(MqttClientDisconnectType disconnectType); Task SubscribeAsync(IList topicFilters); diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index de3a7a48..d5448e80 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -61,21 +61,20 @@ namespace MQTTnet.Server status.ClientId = ClientId; status.IsConnected = _adapter != null; status.Endpoint = _adapter?.Endpoint; - status.ProtocolVersion = _adapter?.PacketSerializer?.ProtocolVersion; + status.ProtocolVersion = _adapter?.PacketSerializerAdapter?.ProtocolVersion; status.PendingApplicationMessagesCount = _pendingPacketsQueue.Count; status.LastPacketReceived = _keepAliveMonitor.LastPacketReceived; status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived; } - public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + public Task RunAsync(MqttApplicationMessage willMessage, int keepAlivePeriod, IMqttChannelAdapter adapter) { - _workerTask = RunInternalAsync(connectPacket, adapter); + _workerTask = RunInternalAsync(willMessage, keepAlivePeriod, adapter); return _workerTask; } - private async Task RunInternalAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + private async Task RunInternalAsync(MqttApplicationMessage willMessage, int keepAlivePeriod, IMqttChannelAdapter adapter) { - if (connectPacket == null) throw new ArgumentNullException(nameof(connectPacket)); if (adapter == null) throw new ArgumentNullException(nameof(adapter)); try @@ -94,10 +93,10 @@ namespace MQTTnet.Server //endworkaround _wasCleanDisconnect = false; - _willMessage = connectPacket.WillMessage; + _willMessage = willMessage; _pendingPacketsQueue.Start(adapter, _cancellationTokenSource.Token); - _keepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token); + _keepAliveMonitor.Start(keepAlivePeriod, _cancellationTokenSource.Token); while (!_cancellationTokenSource.IsCancellationRequested) { diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index d8f4f141..a6a6b9df 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -9,6 +9,7 @@ using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Internal; using MQTTnet.Packets; +using MQTTnet.Packets.V3; using MQTTnet.Protocol; namespace MQTTnet.Server @@ -217,21 +218,18 @@ namespace MQTTnet.Server return; } - if (!(firstPacket is MqttConnectPacket connectPacket)) + if (!(firstPacket is MqttV3ConnectPacket connectPacket)) { throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1]."); } clientId = connectPacket.ClientId; - // Switch to the required protocol version before sending any response. - clientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion; - var connectReturnCode = ValidateConnection(connectPacket, clientAdapter); if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { await clientAdapter.SendPacketAsync( - new MqttConnAckPacket + new MqttV3ConnAckPacket { ConnectReturnCode = connectReturnCode }, @@ -244,7 +242,7 @@ namespace MQTTnet.Server var clientSession = result.Session; await clientAdapter.SendPacketAsync( - new MqttConnAckPacket + new MqttV3ConnAckPacket { ConnectReturnCode = connectReturnCode, IsSessionPresent = result.IsExistingSession @@ -253,7 +251,7 @@ namespace MQTTnet.Server Server.OnClientConnected(clientId); - await clientSession.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false); + await clientSession.RunAsync(connectPacket.WillMessage, connectPacket.KeepAlivePeriod, clientAdapter).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -271,7 +269,7 @@ namespace MQTTnet.Server } } - private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) + private MqttConnectReturnCode ValidateConnection(MqttV3ConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) { if (_options.ConnectionValidator == null) { @@ -289,7 +287,7 @@ namespace MQTTnet.Server return context.ReturnCode; } - private PrepareClientSessionResult PrepareClientSession(MqttConnectPacket connectPacket) + private PrepareClientSessionResult PrepareClientSession(MqttV3ConnectPacket connectPacket) { lock (_sessions) { diff --git a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs index e6cdef23..332f3eaa 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs @@ -1,3 +1,4 @@ +#if NETCOREAPP using System.Buffers; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.AspNetCore; @@ -12,7 +13,7 @@ namespace MQTTnet.AspNetCore.Tests [TestMethod] public void TestTryDeserialize() { - var serializer = new MqttPacketSerializer(); + var serializer = new MqttV311PacketSerializer(); var buffer = serializer.Serialize(new MqttPublishPacket() {Topic = "a", Payload = new byte[5]}); @@ -43,3 +44,4 @@ namespace MQTTnet.AspNetCore.Tests } } } +#endif \ No newline at end of file diff --git a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs index 7f068d60..5ec3c0f6 100644 --- a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs @@ -26,7 +26,8 @@ namespace MQTTnet.Benchmarks .Build(); _packet = message.ToPublishPacket(); - var serializer = new MqttPacketSerializer(); + var serializer = new MqttPacketSerializerAdapter(MqttProtocolVersion.V311); + var serializedPacket = Join(serializer.Serialize(_packet)); _iterations = 10000; diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 9b855805..ed7e5631 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Benchmarks { private MqttBasePacket _packet; private ArraySegment _serializedPacket; - private MqttPacketSerializer _serializer; + private IMqttPacketSerializer _serializer; [GlobalSetup] public void Setup() @@ -29,7 +29,7 @@ namespace MQTTnet.Benchmarks .Build(); _packet = message.ToPublishPacket(); - _serializer = new MqttPacketSerializer(); + _serializer = new MqttV311PacketSerializer(); _serializedPacket = _serializer.Serialize(_packet); } diff --git a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs index 212e1abb..12131a27 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClientTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClientTests.cs @@ -8,6 +8,7 @@ using MQTTnet.Client; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Packets; +using MQTTnet.Packets.V3; namespace MQTTnet.Core.Tests { @@ -50,7 +51,7 @@ namespace MQTTnet.Core.Tests Task.Run(async () => { var connect = await channel2.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None); - await channel2.SendPacketAsync(new MqttConnAckPacket() { ConnectReturnCode = Protocol.MqttConnectReturnCode.ConnectionRefusedNotAuthorized }, CancellationToken.None); + await channel2.SendPacketAsync(new MqttV3ConnAckPacket() { ConnectReturnCode = Protocol.MqttConnectReturnCode.ConnectionRefusedNotAuthorized }, CancellationToken.None); }); diff --git a/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs b/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs index b18904a0..a36555a4 100644 --- a/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttKeepAliveMonitorTests.cs @@ -77,7 +77,7 @@ namespace MQTTnet.Core.Tests throw new NotSupportedException(); } - public Task RunAsync(MqttConnectPacket connectPacket, IMqttChannelAdapter adapter) + public Task RunAsync(MqttApplicationMessage willMessage, int keepAlivePeriod, IMqttChannelAdapter adapter) { throw new NotSupportedException(); } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index c3a75f8e..8f56f0d8 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -7,6 +7,7 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Adapter; using MQTTnet.Internal; using MQTTnet.Packets; +using MQTTnet.Packets.V3; using MQTTnet.Protocol; using MQTTnet.Serializer; @@ -18,7 +19,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV310_MqttConnectPacket() { - var p = new MqttConnectPacket + var p = new MqttV3ConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -33,7 +34,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV311_MqttConnectPacket() { - var p = new MqttConnectPacket + var p = new MqttV3ConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -48,7 +49,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV311_MqttConnectPacketWithWillMessage() { - var p = new MqttConnectPacket + var p = new MqttV3ConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -70,7 +71,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void DeserializeV311_MqttConnectPacket() { - var p = new MqttConnectPacket + var p = new MqttV3ConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -85,7 +86,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void DeserializeV311_MqttConnectPacketWithWillMessage() { - var p = new MqttConnectPacket + var p = new MqttV3ConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -107,7 +108,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV311_MqttConnAckPacket() { - var p = new MqttConnAckPacket + var p = new MqttV3ConnAckPacket { IsSessionPresent = true, ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized @@ -119,7 +120,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV310_MqttConnAckPacket() { - var p = new MqttConnAckPacket + var p = new MqttV3ConnAckPacket { ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized }; @@ -130,7 +131,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void DeserializeV311_MqttConnAckPacket() { - var p = new MqttConnAckPacket + var p = new MqttV3ConnAckPacket { IsSessionPresent = true, ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized @@ -142,7 +143,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void DeserializeV310_MqttConnAckPacket() { - var p = new MqttConnAckPacket + var p = new MqttV3ConnAckPacket { ConnectReturnCode = MqttConnectReturnCode.ConnectionRefusedNotAuthorized }; @@ -153,7 +154,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void Serialize_LargePacket() { - var serializer = new MqttPacketSerializer { ProtocolVersion = MqttProtocolVersion.V311 }; + var serializer = new MqttV311PacketSerializer(); const int payloadLength = 80000; @@ -509,7 +510,20 @@ namespace MQTTnet.Core.Tests private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) { - var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion }; + IMqttPacketSerializer serializer; + if (protocolVersion == MqttProtocolVersion.V311) + { + serializer = new MqttV311PacketSerializer(); + } + else if (protocolVersion == MqttProtocolVersion.V310) + { + serializer = new MqttV310PacketSerializer(); + } + else + { + throw new NotSupportedException(); + } + var data = serializer.Serialize(packet); Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data))); @@ -517,7 +531,19 @@ namespace MQTTnet.Core.Tests private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) { - var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion }; + IMqttPacketSerializer serializer; + if (protocolVersion == MqttProtocolVersion.V311) + { + serializer = new MqttV311PacketSerializer(); + } + else if (protocolVersion == MqttProtocolVersion.V310) + { + serializer = new MqttV310PacketSerializer(); + } + else + { + throw new NotSupportedException(); + } var buffer1 = serializer.Serialize(packet); @@ -541,7 +567,19 @@ namespace MQTTnet.Core.Tests private static T Roundtrip(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) where T : MqttBasePacket { - var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion }; + IMqttPacketSerializer serializer; + if (protocolVersion == MqttProtocolVersion.V311) + { + serializer = new MqttV311PacketSerializer(); + } + else if (protocolVersion == MqttProtocolVersion.V310) + { + serializer = new MqttV310PacketSerializer(); + } + else + { + throw new NotSupportedException(); + } var buffer1 = serializer.Serialize(packet); diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index 0c34bdeb..941e86b2 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -171,14 +171,14 @@ namespace MQTTnet.Core.Tests [TestMethod] public async Task MqttServer_Publish_MultipleClients() { - var serverAdapter = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger()); - var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger()); + var s = new MqttFactory().CreateMqttServer(); var receivedMessagesCount = 0; var locked = new object(); var clientOptions = new MqttClientOptionsBuilder() .WithTcpServer("localhost") .Build(); + var clientOptions2 = new MqttClientOptionsBuilder() .WithTcpServer("localhost") .Build(); diff --git a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs index 9375f07d..1902de4e 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 IMqttPacketSerializer PacketSerializer { get; } = new MqttPacketSerializer(); + public MqttPacketSerializerAdapter PacketSerializerAdapter { get; } = new MqttPacketSerializerAdapter(MqttProtocolVersion.V311); public event EventHandler ReadingPacketStarted; public event EventHandler ReadingPacketCompleted; diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index b557fe02..633f351e 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -14,6 +14,7 @@ using MQTTnet.Extensions.ManagedClient; using MQTTnet.Extensions.Rpc; using MQTTnet.Implementations; using MQTTnet.Protocol; +using MQTTnet.Serializer; using MQTTnet.Server; using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs; using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs; @@ -87,7 +88,8 @@ namespace MQTTnet.TestApp.UniversalWindows var options = new MqttClientOptions { - ClientId = ClientId.Text + ClientId = ClientId.Text, + ProtocolVersion = MqttProtocolVersion.V500 }; if (UseTcp.IsChecked == true)