Add support for V3.1.0; Performance improvements

MQTTnet.Core/Adapter/IMqttCommunicationAdapter.cs

@@ -2,6 +2,7 @@
 using System.Threading.Tasks;
 using MQTTnet.Core.Client;
 using MQTTnet.Core.Packets;
+using MQTTnet.Core.Serializer;
 
 namespace MQTTnet.Core.Adapter
 {
@@ -14,5 +15,7 @@ namespace MQTTnet.Core.Adapter
         Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout);
 
         Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout);
+
+        IMqttPacketSerializer PacketSerializer { get; }
     }
 }

MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs

@@ -11,15 +11,16 @@ namespace MQTTnet.Core.Adapter
 {
     public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
     {
-        private readonly IMqttPacketSerializer _serializer;
         private readonly IMqttCommunicationChannel _channel;
 
         public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer)
         {
             _channel = channel ?? throw new ArgumentNullException(nameof(channel));
-            _serializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
+            PacketSerializer = serializer ?? throw new ArgumentNullException(nameof(serializer));
         }
 
+        public IMqttPacketSerializer PacketSerializer { get; }
+
         public async Task ConnectAsync(MqttClientOptions options, TimeSpan timeout)
         {
             var task = _channel.ConnectAsync(options);
@@ -41,7 +42,7 @@ namespace MQTTnet.Core.Adapter
             bool hasTimeout;
             try
             {
-                var task = _serializer.SerializeAsync(packet, _channel);
+                var task = PacketSerializer.SerializeAsync(packet, _channel);
                 hasTimeout = await Task.WhenAny(Task.Delay(timeout), task) != task;
             }
             catch (Exception exception)
@@ -60,7 +61,7 @@ namespace MQTTnet.Core.Adapter
             MqttBasePacket packet;
             if (timeout > TimeSpan.Zero)
             {
-                var workerTask = _serializer.DeserializeAsync(_channel);
+                var workerTask = PacketSerializer.DeserializeAsync(_channel);
                 var timeoutTask = Task.Delay(timeout);
                 var hasTimeout = Task.WhenAny(timeoutTask, workerTask) == timeoutTask;
 
@@ -73,7 +74,7 @@ namespace MQTTnet.Core.Adapter
             }
             else
             {
-                packet = await _serializer.DeserializeAsync(_channel);
+                packet = await PacketSerializer.DeserializeAsync(_channel);
             }
 
             if (packet == null)

MQTTnet.Core/Client/MqttClient.cs

@@ -27,6 +27,8 @@ namespace MQTTnet.Core.Client
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
             _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter));
+
+            _adapter.PacketSerializer.ProtocolVersion = options.ProtocolVersion;
         }
 
         public event EventHandler Connected;

MQTTnet.Core/Client/MqttClientOptions.cs

@@ -1,4 +1,5 @@
 using System;
+using MQTTnet.Core.Serializer;
 
 namespace MQTTnet.Core.Client
 {
@@ -7,8 +8,8 @@ namespace MQTTnet.Core.Client
         public string Server { get; set; }
 
         public int? Port { get; set; }
-        
-        public MqttClientTlsOptions TlsOptions { get; } = new MqttClientTlsOptions();
+
+        public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
 
         public string UserName { get; set; }
 
@@ -21,5 +22,7 @@ namespace MQTTnet.Core.Client
         public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(5);
 
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
+
+        public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
     }
 }

MQTTnet.Core/Packets/MqttConnectPacket.cs

@@ -1,7 +1,11 @@
-namespace MQTTnet.Core.Packets
+using MQTTnet.Core.Serializer;
+
+namespace MQTTnet.Core.Packets
 {
     public sealed class MqttConnectPacket: MqttBasePacket
     {
+        public MqttProtocolVersion ProtocolVersion { get; set; }
+
         public string ClientId { get; set; }
 
         public string Username { get; set; }

MQTTnet.Core/Serializer/IMqttPacketSerializer.cs

@@ -6,6 +6,8 @@ namespace MQTTnet.Core.Serializer
 {
     public interface IMqttPacketSerializer
     {
+        MqttProtocolVersion ProtocolVersion { get; set; }
+
         Task SerializeAsync(MqttBasePacket mqttPacket, IMqttCommunicationChannel destination);
 
         Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source);

MQTTnet.Core/Serializer/MqttPacketReader.cs

@@ -10,7 +10,7 @@ namespace MQTTnet.Core.Serializer
 {
     public sealed class MqttPacketReader : IDisposable
     {
-        private readonly MemoryStream _remainingData = new MemoryStream();
+        private readonly MemoryStream _remainingData = new MemoryStream(1024);
         private readonly IMqttCommunicationChannel _source;
 
         private int _remainingLength;

MQTTnet.Core/Serializer/MqttPacketSerializer.cs

@@ -9,9 +9,12 @@ using MQTTnet.Core.Protocol;
 
 namespace MQTTnet.Core.Serializer
 {
-    public sealed class MqttV311PacketSerializer : IMqttPacketSerializer
+    public sealed class MqttPacketSerializer : IMqttPacketSerializer
     {
-        private static readonly byte[] MqttV311Prefix = Encoding.UTF8.GetBytes("MQTT");
+        private static byte[] ProtocolVersionV311Name { get; } = Encoding.UTF8.GetBytes("MQTT");
+        private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs");
+
+        public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
 
         public Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
         {
@@ -257,12 +260,21 @@ namespace MQTTnet.Core.Serializer
         private static async Task<MqttBasePacket> DeserializeConnectAsync(MqttPacketReader reader)
         {
             await reader.ReadRemainingDataAsync(2); // Skip 2 bytes
-            
-            var protocolName = await reader.ReadRemainingDataAsync(4);
 
-            if (Encoding.UTF8.GetString(protocolName, 0, protocolName.Length) != "MQTT")
+            MqttProtocolVersion protocolVersion;
+            var protocolName = await reader.ReadRemainingDataAsync(4);
+            if (protocolName.SequenceEqual(ProtocolVersionV310Name))
             {
-                throw new MqttProtocolViolationException("Protocol name is not 'MQTT'.");
+                await reader.ReadRemainingDataAsync(2);
+                protocolVersion = MqttProtocolVersion.V310;
+            }
+            else if (protocolName.SequenceEqual(ProtocolVersionV311Name))
+            {
+                protocolVersion = MqttProtocolVersion.V311;
+            }
+            else
+            {
+                throw new MqttProtocolViolationException("Protocol name is not supported.");
             }
 
             var protocolLevel = await reader.ReadRemainingDataByteAsync();
@@ -273,6 +285,7 @@ namespace MQTTnet.Core.Serializer
 
             var packet = new MqttConnectPacket
             {
+                ProtocolVersion = protocolVersion,
                 CleanSession = connectFlagsReader.Read()
             };
 
@@ -353,7 +366,7 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private static Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttConnectPacket packet, IMqttCommunicationChannel destination)
         {
             ValidateConnectPacket(packet);
 
@@ -361,9 +374,19 @@ namespace MQTTnet.Core.Serializer
             {
                 // Write variable header
                 output.Write(0x00, 0x04); // 3.1.2.1 Protocol Name
-                output.Write(MqttV311Prefix);
-                output.Write(0x04); // 3.1.2.2 Protocol Level
-
+                if (ProtocolVersion == MqttProtocolVersion.V311)
+                {
+                    output.Write(ProtocolVersionV311Name);
+                    output.Write(0x04); // 3.1.2.2 Protocol Level (4)
+                }
+                else
+                {
+                    output.Write(ProtocolVersionV310Name);
+                    output.Write(0x64);
+                    output.Write(0x70);
+                    output.Write(0x03); // Protocol Level (3)
+                }
+                
                 var connectFlags = new ByteWriter(); // 3.1.2.3 Connect Flags
                 connectFlags.Write(false); // Reserved
                 connectFlags.Write(packet.CleanSession);
@@ -408,13 +431,17 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private static Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination)
+        private Task SerializeAsync(MqttConnAckPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())
             {
                 var connectAcknowledgeFlags = new ByteWriter();
-                connectAcknowledgeFlags.Write(packet.IsSessionPresent);
 
+                if (ProtocolVersion == MqttProtocolVersion.V311)
+                {
+                    connectAcknowledgeFlags.Write(packet.IsSessionPresent);
+                }
+                
                 output.Write(connectAcknowledgeFlags);
                 output.Write((byte)packet.ConnectReturnCode);
 
@@ -423,6 +450,17 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
+        private static async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination)
+        {
+            using (var output = new MqttPacketWriter())
+            {
+                output.Write(packet.PacketIdentifier);
+
+                output.InjectFixedHeader(MqttControlPacketType.PubRel, 0x02);
+                await output.WriteToAsync(destination);
+            }
+        }
+
         private static Task SerializeAsync(MqttDisconnectPacket packet, IMqttCommunicationChannel destination)
         {
             return SerializeEmptyPacketAsync(MqttControlPacketType.Disconnect, destination);
@@ -495,17 +533,6 @@ namespace MQTTnet.Core.Serializer
             }
         }
 
-        private static async Task SerializeAsync(MqttPubRelPacket packet, IMqttCommunicationChannel destination)
-        {
-            using (var output = new MqttPacketWriter())
-            {
-                output.Write(packet.PacketIdentifier);
-
-                output.InjectFixedHeader(MqttControlPacketType.PubRel, 0x02);
-                await output.WriteToAsync(destination);
-            }
-        }
-
         private static Task SerializeAsync(MqttPubCompPacket packet, IMqttCommunicationChannel destination)
         {
             using (var output = new MqttPacketWriter())

MQTTnet.Core/Serializer/MqttPacketWriter.cs

@@ -9,13 +9,13 @@ namespace MQTTnet.Core.Serializer
 {
     public sealed class MqttPacketWriter : IDisposable
     {
-        private readonly MemoryStream _buffer = new MemoryStream(512);
+        private readonly MemoryStream _buffer = new MemoryStream(1024);
 
         public void InjectFixedHeader(MqttControlPacketType packetType, byte flags = 0)
         {
-            var fixedHeader = (byte)((byte)packetType << 4);
+            var fixedHeader = (int)packetType << 4;
             fixedHeader |= flags;
-            InjectFixedHeader(fixedHeader);
+            InjectFixedHeader((byte)fixedHeader);
         }
 
         public void Write(byte value)

MQTTnet.Core/Serializer/MqttProtocolVersion.cs

@@ -0,0 +1,8 @@
+namespace MQTTnet.Core.Serializer
+{
+    public enum MqttProtocolVersion
+    {
+        V311,
+        V310
+    }
+}

MQTTnet.Core/Server/ConnectedMqttClient.cs

@@ -0,0 +1,11 @@
+using MQTTnet.Core.Serializer;
+
+namespace MQTTnet.Core.Server
+{
+    public class ConnectedMqttClient
+    {
+        public string ClientId { get; set; }
+
+        public MqttProtocolVersion ProtocolVersion { get; set; }
+    }
+}

MQTTnet.Core/Server/IMqttServer.cs

@@ -9,7 +9,7 @@ namespace MQTTnet.Core.Server
         event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
         event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
 
-        IList<string> GetConnectedClients();
+        IList<ConnectedMqttClient> GetConnectedClients();
         void InjectClient(string identifier, IMqttCommunicationAdapter adapter);
         void Publish(MqttApplicationMessage applicationMessage);
         void Start();

MQTTnet.Core/Server/MqttClientSession.cs

@@ -21,7 +21,6 @@ namespace MQTTnet.Core.Server
         private readonly MqttServerOptions _options;
 
         private CancellationTokenSource _cancellationTokenSource;
-        private IMqttCommunicationAdapter _adapter;
         private string _identifier;
         private MqttApplicationMessage _willApplicationMessage;
 
@@ -36,7 +35,9 @@ namespace MQTTnet.Core.Server
 
         public string ClientId { get; }
 
-        public bool IsConnected => _adapter != null;
+        public bool IsConnected => Adapter != null;
+
+        public IMqttCommunicationAdapter Adapter { get; private set; }
 
         public async Task RunAsync(string identifier, MqttApplicationMessage willApplicationMessage, IMqttCommunicationAdapter adapter)
         {
@@ -47,7 +48,7 @@ namespace MQTTnet.Core.Server
             try
             {
                 _identifier = identifier;
-                _adapter = adapter;
+                Adapter = adapter;
                 _cancellationTokenSource = new CancellationTokenSource();
 
                 _messageQueue.Start(adapter);
@@ -73,7 +74,7 @@ namespace MQTTnet.Core.Server
 
                 _messageQueue.Stop();
                 _cancellationTokenSource.Cancel();
-                _adapter = null;
+                Adapter = null;
 
                 MqttTrace.Information(nameof(MqttClientSession), $"Client '{_identifier}': Disconnected.");
             }
@@ -102,12 +103,12 @@ namespace MQTTnet.Core.Server
         {
             if (packet is MqttSubscribePacket subscribePacket)
             {
-                return _adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
+                return Adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
             }
 
             if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
-                return _adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
+                return Adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
             }
 
             if (packet is MqttPublishPacket publishPacket)
@@ -122,7 +123,7 @@ namespace MQTTnet.Core.Server
 
             if (packet is MqttPubRecPacket pubRecPacket)
             {
-                return _adapter.SendPacketAsync(pubRecPacket.CreateResponse<MqttPubRelPacket>(), _options.DefaultCommunicationTimeout);
+                return Adapter.SendPacketAsync(pubRecPacket.CreateResponse<MqttPubRelPacket>(), _options.DefaultCommunicationTimeout);
             }
 
             if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
@@ -133,7 +134,7 @@ namespace MQTTnet.Core.Server
 
             if (packet is MqttPingReqPacket)
             {
-                return _adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
+                return Adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
             }
 
             if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
@@ -159,7 +160,7 @@ namespace MQTTnet.Core.Server
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
                 _publishPacketReceivedCallback(this, publishPacket);
-                return _adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
+                return Adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
 
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -172,7 +173,7 @@ namespace MQTTnet.Core.Server
 
                 _publishPacketReceivedCallback(this, publishPacket);
 
-                return _adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
+                return Adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
 
             throw new MqttCommunicationException("Received a not supported QoS level.");
@@ -185,7 +186,7 @@ namespace MQTTnet.Core.Server
                 _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
             }
 
-            return _adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
+            return Adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
         }
     }
 }

MQTTnet.Core/Server/MqttClientSessionsManager.cs

@@ -34,6 +34,9 @@ namespace MQTTnet.Core.Server
                     throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
                 }
 
+                // Switch to the required protocol version before sending any response.
+                eventArgs.ClientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion;
+
                 var connectReturnCode = ValidateConnection(connectPacket);
                 if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
                 {
@@ -73,11 +76,15 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        public IList<string> GetConnectedClients()
+        public IList<ConnectedMqttClient> GetConnectedClients()
         {
             lock (_syncRoot)
             {
-                return _clientSessions.Where(s => s.Value.IsConnected).Select(s => s.Key).ToList();
+                return _clientSessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
+                {
+                    ClientId = s.Value.ClientId,
+                    ProtocolVersion = s.Value.Adapter.PacketSerializer.ProtocolVersion
+                }).ToList();
             }
         }
 

MQTTnet.Core/Server/MqttServer.cs

@@ -25,7 +25,7 @@ namespace MQTTnet.Core.Server
             _clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
         }
 
-        public IList<string> GetConnectedClients()
+        public IList<ConnectedMqttClient> GetConnectedClients()
         {
             return _clientSessionsManager.GetConnectedClients();
         }