Add interfaces, optimize performance, fix QoS level 2

MQTTnet.Core/Server/IMqttServer.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Collections.Generic;
+using MQTTnet.Core.Adapter;
+
+namespace MQTTnet.Core.Server
+{
+    public interface IMqttServer
+    {
+        event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+        event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+
+        IList<string> GetConnectedClients();
+        void InjectClient(string identifier, IMqttCommunicationAdapter adapter);
+        void Publish(MqttApplicationMessage applicationMessage);
+        void Start();
+        void Stop();
+    }
+}

MQTTnet.Core/Server/MqttClientMessageQueue.cs

@@ -35,7 +35,7 @@ namespace MQTTnet.Core.Server
             _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter));
             _cancellationTokenSource = new CancellationTokenSource();
 
-            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token));
+            Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
         }
 
         public void Stop()
@@ -45,14 +45,13 @@ namespace MQTTnet.Core.Server
             _cancellationTokenSource = null;
         }
 
-        public void Enqueue(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public void Enqueue(MqttPublishPacket publishPacket)
         {
-            if (senderClientSession == null) throw new ArgumentNullException(nameof(senderClientSession));
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
             lock (_pendingPublishPackets)
             {
-                _pendingPublishPackets.Add(new MqttClientPublishPacketContext(senderClientSession, publishPacket));
+                _pendingPublishPackets.Add(new MqttClientPublishPacketContext(publishPacket));
                 _gate.Set();
             }
         }

MQTTnet.Core/Server/MqttClientPublishPacketContext.cs

@@ -5,14 +5,11 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientPublishPacketContext
     {
-        public MqttClientPublishPacketContext(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public MqttClientPublishPacketContext(MqttPublishPacket publishPacket)
         {
-            SenderClientSession = senderClientSession ?? throw new ArgumentNullException(nameof(senderClientSession));
             PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
         }
 
-        public MqttClientSession SenderClientSession { get; }
-
         public MqttPublishPacket PublishPacket { get; }
 
         public int SendTries { get; set; }

MQTTnet.Core/Server/MqttClientSession.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
@@ -13,13 +13,13 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSession : IDisposable
     {
-        private readonly ConcurrentDictionary<ushort, MqttPublishPacket> _pendingIncomingPublications = new ConcurrentDictionary<ushort, MqttPublishPacket>();
+        private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
 
         private readonly MqttClientSubscriptionsManager _subscriptionsManager = new MqttClientSubscriptionsManager();
         private readonly MqttClientMessageQueue _messageQueue;
         private readonly Action<MqttClientSession, MqttPublishPacket> _publishPacketReceivedCallback;
         private readonly MqttServerOptions _options;
-        
+
         private CancellationTokenSource _cancellationTokenSource;
         private IMqttCommunicationAdapter _adapter;
         private string _identifier;
@@ -79,17 +79,16 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        public void EnqueuePublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public void EnqueuePublishPacket(MqttPublishPacket publishPacket)
         {
-            if (senderClientSession == null) throw new ArgumentNullException(nameof(senderClientSession));
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            if (!_subscriptionsManager.IsTopicSubscribed(publishPacket))
+            if (!_subscriptionsManager.IsSubscribed(publishPacket))
             {
                 return;
             }
 
-            _messageQueue.Enqueue(senderClientSession, publishPacket);
+            _messageQueue.Enqueue(publishPacket);
             MqttTrace.Verbose(nameof(MqttClientSession), $"Client '{_identifier}: Enqueued pending publish packet.");
         }
 
@@ -101,34 +100,35 @@ namespace MQTTnet.Core.Server
 
         private Task HandleIncomingPacketAsync(MqttBasePacket packet)
         {
-            var subscribePacket = packet as MqttSubscribePacket;
-            if (subscribePacket != null)
+            if (packet is MqttSubscribePacket subscribePacket)
             {
                 return _adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
             }
 
-            var unsubscribePacket = packet as MqttUnsubscribePacket;
-            if (unsubscribePacket != null)
+            if (packet is MqttUnsubscribePacket unsubscribePacket)
             {
                 return _adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
             }
 
-            var publishPacket = packet as MqttPublishPacket;
-            if (publishPacket != null)
+            if (packet is MqttPublishPacket publishPacket)
             {
                 return HandleIncomingPublishPacketAsync(publishPacket);
             }
 
-            var pubRelPacket = packet as MqttPubRelPacket;
-            if (pubRelPacket != null)
+            if (packet is MqttPubRelPacket pubRelPacket)
             {
                 return HandleIncomingPubRelPacketAsync(pubRelPacket);
             }
 
-            var pubAckPacket = packet as MqttPubAckPacket;
-            if (pubAckPacket != null)
+            if (packet is MqttPubRecPacket pubRecPacket)
             {
-                return HandleIncomingPubAckPacketAsync(pubAckPacket);
+                return _adapter.SendPacketAsync(pubRecPacket.CreateResponse<MqttPubRelPacket>(), _options.DefaultCommunicationTimeout);
+            }
+
+            if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
+            {
+                // Discard message.
+                return Task.FromResult((object)null);
             }
 
             if (packet is MqttPingReqPacket)
@@ -148,12 +148,7 @@ namespace MQTTnet.Core.Server
             return Task.FromResult((object)null);
         }
 
-        private async Task HandleIncomingPubAckPacketAsync(MqttPubAckPacket pubAckPacket)
-        {
-            await Task.FromResult((object)null);
-        }
-
-        private async Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
+        private Task HandleIncomingPublishPacketAsync(MqttPublishPacket publishPacket)
         {
             if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
@@ -161,26 +156,33 @@ namespace MQTTnet.Core.Server
             }
             else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
-                await _adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
                 _publishPacketReceivedCallback(this, publishPacket);
+                return _adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
             else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
             {
-                _pendingIncomingPublications[publishPacket.PacketIdentifier] = publishPacket;
-                await _adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
+                // QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
+                lock (_unacknowledgedPublishPackets)
+                {
+                    _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
+                }
+
+                _publishPacketReceivedCallback(this, publishPacket);
+
+                return _adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
             }
+
+            throw new MqttCommunicationException("Received not supported QoS level.");
         }
 
-        private async Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket)
+        private Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket)
         {
-            MqttPublishPacket publishPacket;
-            if (!_pendingIncomingPublications.TryRemove(pubRelPacket.PacketIdentifier, out publishPacket))
+            lock (_unacknowledgedPublishPackets)
             {
-                return;
+                _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
             }
 
-            await _adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
-            _publishPacketReceivedCallback(this, publishPacket);
+            return _adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
         }
     }
 }

MQTTnet.Core/Server/MqttClientSessionsManager.cs

@@ -22,7 +22,7 @@ namespace MQTTnet.Core.Server
             _options = options ?? throw new ArgumentNullException(nameof(options));
         }
 
-        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; 
+        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
         public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
         {
@@ -127,14 +127,24 @@ namespace MQTTnet.Core.Server
             }
         }
 
-        private void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
+        public void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
         {
-            var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession.ClientId, publishPacket.ToApplicationMessage());
-            ApplicationMessageReceived?.Invoke(this, eventArgs);
-
-            foreach (var clientSession in _clientSessions.Values.ToList())
+            try
             {
-                clientSession.EnqueuePublishPacket(senderClientSession, publishPacket);
+                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, publishPacket.ToApplicationMessage());
+                ApplicationMessageReceived?.Invoke(this, eventArgs);
+            }
+            catch (Exception exception)
+            {
+                MqttTrace.Error(nameof(MqttClientSessionsManager), exception, "Error while processing application message");
+            }
+
+            lock (_syncRoot)
+            {
+                foreach (var clientSession in _clientSessions.Values.ToList())
+                {
+                    clientSession.EnqueuePublishPacket(publishPacket);
+                }
             }
         }
     }

MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs

@@ -1,5 +1,5 @@
 using System;
-using System.Collections.Concurrent;
+using System.Collections.Generic;
 using MQTTnet.Core.Packets;
 using MQTTnet.Core.Protocol;
 
@@ -7,17 +7,21 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttClientSubscriptionsManager
     {
-        private readonly ConcurrentDictionary<string, MqttQualityOfServiceLevel> _subscribedTopics = new ConcurrentDictionary<string, MqttQualityOfServiceLevel>();
+        private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscribedTopics = new Dictionary<string, MqttQualityOfServiceLevel>();
 
         public MqttSubAckPacket Subscribe(MqttSubscribePacket subscribePacket)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
 
             var responsePacket = subscribePacket.CreateResponse<MqttSubAckPacket>();
-            foreach (var topicFilter in subscribePacket.TopicFilters)
+
+            lock (_subscribedTopics)
             {
-                _subscribedTopics[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
-                responsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1); // TODO: Add support for QoS 2.
+                foreach (var topicFilter in subscribePacket.TopicFilters)
+                {
+                    _subscribedTopics[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
+                    responsePacket.SubscribeReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1); // TODO: Add support for QoS 2.
+                }
             }
 
             return responsePacket;
@@ -27,32 +31,37 @@ namespace MQTTnet.Core.Server
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
 
-            foreach (var topicFilter in unsubscribePacket.TopicFilters)
+            lock (_subscribedTopics)
             {
-                MqttQualityOfServiceLevel _;
-                _subscribedTopics.TryRemove(topicFilter, out _);
+                foreach (var topicFilter in unsubscribePacket.TopicFilters)
+                {
+                    _subscribedTopics.Remove(topicFilter);
+                }
             }
 
             return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>();
         }
 
-        public bool IsTopicSubscribed(MqttPublishPacket publishPacket)
+        public bool IsSubscribed(MqttPublishPacket publishPacket)
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
-            foreach (var subscribedTopic in _subscribedTopics)
+            lock (_subscribedTopics)
             {
-                if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscribedTopic.Key))
+                foreach (var subscribedTopic in _subscribedTopics)
                 {
-                    continue;
-                }
+                    if (publishPacket.QualityOfServiceLevel > subscribedTopic.Value)
+                    {
+                        continue;
+                    }
 
-                if (subscribedTopic.Value < publishPacket.QualityOfServiceLevel)
-                {
-                    continue;
-                }
+                    if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscribedTopic.Key))
+                    {
+                        continue;
+                    }
 
-                return true;
+                    return true;
+                }
             }
 
             return false;

MQTTnet.Core/Server/MqttServer.cs

@@ -4,10 +4,11 @@ using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Diagnostics;
+using MQTTnet.Core.Internal;
 
 namespace MQTTnet.Core.Server
 {
-    public sealed class MqttServer
+    public sealed class MqttServer : IMqttServer
     {
         private readonly MqttClientSessionsManager _clientSessionsManager;
         private readonly ICollection<IMqttServerAdapter> _adapters;
@@ -33,6 +34,13 @@ namespace MQTTnet.Core.Server
 
         public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
 
+        public void Publish(MqttApplicationMessage applicationMessage)
+        {
+            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+
+            _clientSessionsManager.DispatchPublishPacket(null, applicationMessage.ToPublishPacket());
+        }
+
         public void InjectClient(string identifier, IMqttCommunicationAdapter adapter)
         {
             if (adapter == null) throw new ArgumentNullException(nameof(adapter));

MQTTnet.Core/Server/MqttServerDefaultEndpointOptions.cs

@@ -1,6 +1,6 @@
 namespace MQTTnet.Core.Server
 {
-    public sealed class DefaultEndpointOptions
+    public sealed class MqttServerDefaultEndpointOptions
     {
         public bool IsEnabled { get; set; } = true;
 

MQTTnet.Core/Server/MqttServerOptions.cs

@@ -6,13 +6,13 @@ namespace MQTTnet.Core.Server
 {
     public sealed class MqttServerOptions
     {
-        public DefaultEndpointOptions DefaultEndpointOptions { get; } = new DefaultEndpointOptions();
+        public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions();
 
         public MqttServerTlsEndpointOptions TlsEndpointOptions { get; } = new MqttServerTlsEndpointOptions();
         
         public int ConnectionBacklog { get; set; } = 10;
 
-        public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10);
+        public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
 
         public Func<MqttConnectPacket, MqttConnectReturnCode> ConnectionValidator { get; set; }
     }

MQTTnet.Core/Server/MqttServerTlsEndpointOptionsExtensions.cs

@@ -1,19 +0,0 @@
-using System;
-
-namespace MQTTnet.Core.Server
-{
-    public static class MqttServerTlsEndpointOptionsExtensions
-    {
-        public static int GetPort(this DefaultEndpointOptions options)
-        {
-            if (options == null) throw new ArgumentNullException(nameof(options));
-
-            if (!options.Port.HasValue)
-            {
-                return 1883;
-            }
-
-            return options.Port.Value;
-        }
-    }
-}