diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index ce91a487..9e073cbb 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -196,7 +196,7 @@ namespace MQTTnet.Client private async Task AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken) { - var connectPacket = new MqttV3ConnectPacket + var connectPacket = new MqttConnectPacket { ClientId = _options.ClientId, Username = _options.Credentials?.Username, diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 5c9ba727..68099169 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -64,7 +64,9 @@ - + + + \ No newline at end of file diff --git a/Source/MQTTnet/Packets/MqttConnectPacket.cs b/Source/MQTTnet/Packets/MqttConnectPacket.cs index 1faa8c20..7c0b7a00 100644 --- a/Source/MQTTnet/Packets/MqttConnectPacket.cs +++ b/Source/MQTTnet/Packets/MqttConnectPacket.cs @@ -1,6 +1,34 @@ -namespace MQTTnet.Packets +using System.Collections.Generic; +using MQTTnet.Packets.Properties; + +namespace MQTTnet.Packets { - public abstract class MqttConnectPacket : MqttBasePacket + public class MqttConnectPacket : MqttBasePacket { + 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; } + + /// + /// MQTTv5: Also called "Clean Start". + /// + public bool CleanSession { get; set; } + + public MqttApplicationMessage WillMessage { get; set; } + + public List Properties { 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/Packets/Properties/ByteValue.cs b/Source/MQTTnet/Packets/Properties/ByteValue.cs new file mode 100644 index 00000000..5c34cb90 --- /dev/null +++ b/Source/MQTTnet/Packets/Properties/ByteValue.cs @@ -0,0 +1,22 @@ +using System; +using MQTTnet.Serializer; + +namespace MQTTnet.Packets.Properties +{ + public class ByteValue : IPropertyValue + { + public ByteValue(byte value) + { + Value = value; + } + + public byte Value { get; } + + public void WriteTo(MqttPacketWriter writer) + { + if (writer == null) throw new ArgumentNullException(nameof(writer)); + + writer.Write(Value); + } + } +} diff --git a/Source/MQTTnet/Packets/Properties/IPropertyValue.cs b/Source/MQTTnet/Packets/Properties/IPropertyValue.cs new file mode 100644 index 00000000..005134fc --- /dev/null +++ b/Source/MQTTnet/Packets/Properties/IPropertyValue.cs @@ -0,0 +1,9 @@ +using MQTTnet.Serializer; + +namespace MQTTnet.Packets.Properties +{ + public interface IPropertyValue + { + void WriteTo(MqttPacketWriter writer); + } +} diff --git a/Source/MQTTnet/Packets/Properties/Property.cs b/Source/MQTTnet/Packets/Properties/Property.cs new file mode 100644 index 00000000..1fc71d6d --- /dev/null +++ b/Source/MQTTnet/Packets/Properties/Property.cs @@ -0,0 +1,9 @@ +namespace MQTTnet.Packets.Properties +{ + public class Property + { + public PropertyType Type { get; set; } + + public IPropertyValue Value { get; set; } + } +} diff --git a/Source/MQTTnet/Packets/Properties/PropertyType.cs b/Source/MQTTnet/Packets/Properties/PropertyType.cs new file mode 100644 index 00000000..8d60aad7 --- /dev/null +++ b/Source/MQTTnet/Packets/Properties/PropertyType.cs @@ -0,0 +1,33 @@ +namespace MQTTnet.Packets.Properties +{ + public enum PropertyType + { + PayloadFormatIndicator = 1, + MessageExpiryInterval = 2, + ContentType = 3, + ResponseTopic = 8, + CorrelationData = 9, + SubscriptionIdentifier = 11, + SessionExpiryInterval = 17, + AssignedClientIdentifer = 18, + ServerKeepAlive = 19, + AuthenticationMethod = 21, + AuthenticationData = 22, + RequestProblemInformation = 23, + WillDelayInterval = 24, + RequestResponseInformation = 25, + ResponseInformation = 26, + ServerReference = 28, + ReasonString = 31, + ReceiveMaximum = 33, + TopicAliasMaximum = 34, + TopicAlias = 35, + MaximumQoS = 36, + RetainAvailable = 37, + UserProperty = 38, + MaximumPacketSize = 39, + WildcardSubscriptionAvailable = 40, + SubscriptionIdentifierAvailable = 41, + SharedSubscriptionAvailable = 42 + } +} diff --git a/Source/MQTTnet/Packets/Properties/StringValue.cs b/Source/MQTTnet/Packets/Properties/StringValue.cs new file mode 100644 index 00000000..fc614ad1 --- /dev/null +++ b/Source/MQTTnet/Packets/Properties/StringValue.cs @@ -0,0 +1,22 @@ +using System; +using MQTTnet.Serializer; + +namespace MQTTnet.Packets.Properties +{ + public class StringValue : IPropertyValue + { + public StringValue(string value) + { + Value = value; + } + + public string Value { get; } + + public void WriteTo(MqttPacketWriter writer) + { + if (writer == null) throw new ArgumentNullException(nameof(writer)); + + writer.WriteWithLengthPrefix(Value); + } + } +} diff --git a/Source/MQTTnet/Packets/Properties/TwoByteIntegerValue.cs b/Source/MQTTnet/Packets/Properties/TwoByteIntegerValue.cs new file mode 100644 index 00000000..79fc919c --- /dev/null +++ b/Source/MQTTnet/Packets/Properties/TwoByteIntegerValue.cs @@ -0,0 +1,22 @@ +using System; +using MQTTnet.Serializer; + +namespace MQTTnet.Packets.Properties +{ + public class TwoByteIntegerValue : IPropertyValue + { + public TwoByteIntegerValue(ushort value) + { + Value = value; + } + + public ushort Value { get; } + + public void WriteTo(MqttPacketWriter writer) + { + if (writer == null) throw new ArgumentNullException(nameof(writer)); + + throw new NotImplementedException(); + } + } +} diff --git a/Source/MQTTnet/Packets/V3/MqttV3ConnectPacket.cs b/Source/MQTTnet/Packets/V3/MqttV3ConnectPacket.cs deleted file mode 100644 index 2434d594..00000000 --- a/Source/MQTTnet/Packets/V3/MqttV3ConnectPacket.cs +++ /dev/null @@ -1,26 +0,0 @@ -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/MqttPacketSerializerAdapter.cs b/Source/MQTTnet/Serializer/MqttPacketSerializerAdapter.cs index 9a681e90..4a4a164a 100644 --- a/Source/MQTTnet/Serializer/MqttPacketSerializerAdapter.cs +++ b/Source/MQTTnet/Serializer/MqttPacketSerializerAdapter.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Serializer { if (Serializer == null) { - throw new InvalidOperationException("Protocol version nor set or detected."); + throw new InvalidOperationException("Protocol version not set or detected."); } return Serializer.Deserialize(receivedMqttPacket); @@ -48,6 +48,10 @@ namespace MQTTnet.Serializer public void DetectProtocolVersion(ReceivedMqttPacket receivedMqttPacket) { var protocolVersion = ParseProtocolVersion(receivedMqttPacket); + + // Reset the position of the stream beacuse the protocol version is part of + // the regular CONNECT packet. So it will not properly deserialized if this + // data is missing. receivedMqttPacket.Body.Seek(0); UseProtocolVersion(protocolVersion); diff --git a/Source/MQTTnet/Serializer/MqttPacketWriter.cs b/Source/MQTTnet/Serializer/MqttPacketWriter.cs index 10ec4f43..beb146fb 100644 --- a/Source/MQTTnet/Serializer/MqttPacketWriter.cs +++ b/Source/MQTTnet/Serializer/MqttPacketWriter.cs @@ -27,9 +27,9 @@ namespace MQTTnet.Serializer return (byte)fixedHeader; } - public static ArraySegment EncodeRemainingLength(int length) + public static ArraySegment EncodeVariableByteInteger(int value) { - if (length <= 0) + if (value <= 0) { return new ArraySegment(new byte[1], 0, 1); } @@ -37,7 +37,7 @@ namespace MQTTnet.Serializer var buffer = new byte[4]; var bufferOffset = 0; - var x = length; + var x = value; do { var encodedByte = x % 128; @@ -54,6 +54,11 @@ namespace MQTTnet.Serializer return new ArraySegment(buffer, 0, bufferOffset); } + public void WriteVariableLengthInteger(int value) + { + Write(EncodeVariableByteInteger(value)); + } + public void WriteWithLengthPrefix(string value) { WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty)); @@ -85,16 +90,30 @@ namespace MQTTnet.Serializer IncreasePostition(1); } - public void Write(byte[] array, int offset, int count) + public void Write(byte[] buffer, int offset, int count) { - if (array == null) throw new ArgumentNullException(nameof(array)); + if (buffer == null) throw new ArgumentNullException(nameof(buffer)); EnsureAdditionalCapacity(count); - Array.Copy(array, offset, _buffer, _position, count); + Array.Copy(buffer, offset, _buffer, _position, count); IncreasePostition(count); } + public void Write(ArraySegment buffer) + { + if (buffer == null) throw new ArgumentNullException(nameof(buffer)); + + Write(buffer.Array, buffer.Offset, buffer.Count); + } + + public void Write(MqttPacketWriter propertyWriter) + { + if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter)); + + Write(propertyWriter._buffer, 0, propertyWriter.Length); + } + public void Reset() { Length = 5; diff --git a/Source/MQTTnet/Serializer/MqttV310PacketSerializer.cs b/Source/MQTTnet/Serializer/MqttV310PacketSerializer.cs index 572feb9b..bb72db14 100644 --- a/Source/MQTTnet/Serializer/MqttV310PacketSerializer.cs +++ b/Source/MQTTnet/Serializer/MqttV310PacketSerializer.cs @@ -25,7 +25,7 @@ namespace MQTTnet.Serializer var fixedHeader = SerializePacket(packet, _packetWriter); var remainingLength = _packetWriter.Length - 5; - var remainingLengthBuffer = MqttPacketWriter.EncodeRemainingLength(remainingLength); + var remainingLengthBuffer = MqttPacketWriter.EncodeVariableByteInteger(remainingLength); var headerSize = FixedHeaderSize + remainingLengthBuffer.Count; var headerOffset = 5 - headerSize; @@ -79,7 +79,7 @@ namespace MQTTnet.Serializer { switch (packet) { - case MqttV3ConnectPacket connectPacket: return SerializeConnectPacket(connectPacket, packetWriter); + case MqttConnectPacket connectPacket: return SerializeConnectPacket(connectPacket, packetWriter); case MqttV3ConnAckPacket connAckPacket: return SerializeConnAckPacket(connAckPacket, packetWriter); case MqttDisconnectPacket _: return SerializeEmptyPacket(MqttControlPacketType.Disconnect); case MqttPingReqPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingReq); @@ -216,7 +216,7 @@ namespace MQTTnet.Serializer { ThrowIfBodyIsEmpty(body); - var packet = new MqttV3ConnectPacket + var packet = new MqttConnectPacket { ProtocolName = body.ReadStringWithLengthPrefix(), ProtocolLevel = body.ReadByte() @@ -293,7 +293,7 @@ namespace MQTTnet.Serializer return packet; } - protected void ValidateConnectPacket(MqttV3ConnectPacket packet) + protected void ValidateConnectPacket(MqttConnectPacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); @@ -301,6 +301,11 @@ namespace MQTTnet.Serializer { throw new MqttProtocolViolationException("CleanSession must be set if ClientId is empty [MQTT-3.1.3-7]."); } + + if (packet.Properties != null) + { + throw new MqttProtocolViolationException("Properties are only supported in MQTT Version 5."); + } } // ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local @@ -312,7 +317,7 @@ namespace MQTTnet.Serializer } } - protected virtual byte SerializeConnectPacket(MqttV3ConnectPacket packet, MqttPacketWriter packetWriter) + protected virtual byte SerializeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) { ValidateConnectPacket(packet); diff --git a/Source/MQTTnet/Serializer/MqttV311PacketSerializer.cs b/Source/MQTTnet/Serializer/MqttV311PacketSerializer.cs index f68ae02f..f124c7d1 100644 --- a/Source/MQTTnet/Serializer/MqttV311PacketSerializer.cs +++ b/Source/MQTTnet/Serializer/MqttV311PacketSerializer.cs @@ -7,7 +7,7 @@ namespace MQTTnet.Serializer { public class MqttV311PacketSerializer : MqttV310PacketSerializer { - protected override byte SerializeConnectPacket(MqttV3ConnectPacket packet, MqttPacketWriter packetWriter) + protected override byte SerializeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) { ValidateConnectPacket(packet); diff --git a/Source/MQTTnet/Serializer/MqttV500PacketSerializer.cs b/Source/MQTTnet/Serializer/MqttV500PacketSerializer.cs index a50a9033..44ec36be 100644 --- a/Source/MQTTnet/Serializer/MqttV500PacketSerializer.cs +++ b/Source/MQTTnet/Serializer/MqttV500PacketSerializer.cs @@ -1,24 +1,96 @@ using System; using MQTTnet.Adapter; +using MQTTnet.Exceptions; using MQTTnet.Packets; +using MQTTnet.Protocol; namespace MQTTnet.Serializer { - public class MqttV500PacketSerializer : IMqttPacketSerializer + public class MqttV500PacketSerializer : MqttV311PacketSerializer { - public ArraySegment Serialize(MqttBasePacket mqttPacket) - { - throw new NotImplementedException(); - } + //public ArraySegment Serialize(MqttBasePacket mqttPacket) + //{ + // throw new NotImplementedException(); + //} - public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket) - { - throw new NotImplementedException(); - } + //public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket) + //{ + // throw new NotImplementedException(); + //} - public void FreeBuffer() + protected override byte SerializeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter) { - throw new NotImplementedException(); + 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); + } + + if (packet.Properties != null) + { + var propertyWriter = new MqttPacketWriter(); + foreach (var property in packet.Properties) + { + propertyWriter.Write((byte)property.Key); + property.Value.WriteTo(propertyWriter); + } + + packetWriter.WriteVariableLengthInteger(propertyWriter.Length); + packetWriter.Write(propertyWriter); + } + + return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect); } } } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index f180f685..a06e969e 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -218,7 +218,7 @@ namespace MQTTnet.Server return; } - if (!(firstPacket is MqttV3ConnectPacket connectPacket)) + if (!(firstPacket is MqttConnectPacket connectPacket)) { throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1]."); } @@ -269,7 +269,7 @@ namespace MQTTnet.Server } } - private MqttConnectReturnCode ValidateConnection(MqttV3ConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) + private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) { if (_options.ConnectionValidator == null) { @@ -287,7 +287,7 @@ namespace MQTTnet.Server return context.ReturnCode; } - private PrepareClientSessionResult PrepareClientSession(MqttV3ConnectPacket connectPacket) + private PrepareClientSessionResult PrepareClientSession(MqttConnectPacket connectPacket) { lock (_sessions) { diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index 8f56f0d8..5df4b69b 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV310_MqttConnectPacket() { - var p = new MqttV3ConnectPacket + var p = new MqttConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -34,7 +34,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV311_MqttConnectPacket() { - var p = new MqttV3ConnectPacket + var p = new MqttConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -49,7 +49,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void SerializeV311_MqttConnectPacketWithWillMessage() { - var p = new MqttV3ConnectPacket + var p = new MqttConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -71,7 +71,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void DeserializeV311_MqttConnectPacket() { - var p = new MqttV3ConnectPacket + var p = new MqttConnectPacket { ClientId = "XYZ", Password = "PASS", @@ -86,7 +86,7 @@ namespace MQTTnet.Core.Tests [TestMethod] public void DeserializeV311_MqttConnectPacketWithWillMessage() { - var p = new MqttV3ConnectPacket + var p = new MqttConnectPacket { ClientId = "XYZ", Password = "PASS", diff --git a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs index c405f5fe..a8e4ce93 100644 --- a/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs +++ b/Tests/MQTTnet.Core.Tests/RoundtripTimeTests.cs @@ -36,6 +36,8 @@ namespace MQTTnet.Core.Tests var times = new List(); var stopwatch = Stopwatch.StartNew(); + await Task.Delay(1000); + for (var i = 0; i < 100; i++) { response = new TaskCompletionSource();