Wrap COM exceptions, close client connection when server stops, process retained messages from server

2026-10-19 19:13:11 +08:00 · 2017-11-16 00:28:56 +01:00
parent b956623234
commit 8de87c8772
10 changed files with 204 additions and 228 deletions
@@ -10,8 +10,7 @@ namespace MQTTnet.Core.Server
        event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
        event EventHandler<MqttServerStartedEventArgs> Started;

-        IList<ConnectedMqttClient> GetConnectedClients();
-        void Publish(IEnumerable<MqttApplicationMessage> applicationMessages);
+        Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();

        Task StartAsync();
        Task StopAsync();
@@ -96,7 +96,7 @@ namespace MQTTnet.Core.Server
                    _pendingPublishPackets.Add(packet, CancellationToken.None);
                }

-                _session.Stop();
+                await _session.StopAsync();
            }
        }
    }
@@ -17,7 +17,6 @@ namespace MQTTnet.Core.Server
    {
        private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();

-        private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;
        private readonly MqttClientSubscriptionsManager _subscriptionsManager;
        private readonly MqttClientSessionsManager _sessionsManager;
        private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
@@ -34,10 +33,8 @@ namespace MQTTnet.Core.Server
            MqttClientSessionsManager sessionsManager,
            MqttClientSubscriptionsManager subscriptionsManager,
            ILogger<MqttClientSession> logger,
-            ILogger<MqttClientPendingMessagesQueue> messageQueueLogger,
-            IMqttClientRetainedMessageManager clientRetainedMessageManager)
+            ILogger<MqttClientPendingMessagesQueue> messageQueueLogger)
        {
-            _clientRetainedMessageManager = clientRetainedMessageManager ?? throw new ArgumentNullException(nameof(clientRetainedMessageManager));
            _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
            _subscriptionsManager = subscriptionsManager ?? throw new ArgumentNullException(nameof(subscriptionsManager));
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -82,7 +79,7 @@ namespace MQTTnet.Core.Server
            }
        }

-        public void Stop()
+        public async Task StopAsync()
        {
            try
            {
@@ -90,17 +87,21 @@ namespace MQTTnet.Core.Server
                _cancellationTokenSource?.Dispose();
                _cancellationTokenSource = null;

-                _adapter = null;
+                if (_adapter != null)
+                {
+                    await _adapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
+                    _adapter = null;
+                }

-                _logger.LogInformation("Client '{0}': Disconnected.", ClientId);
+                _logger.LogInformation("Client '{0}': Session stopped.", ClientId);
            }
            finally
            {
-                var willMessage = _willMessage; 
+                var willMessage = _willMessage;
                if (willMessage != null)
                {
                    _willMessage = null; //clear willmessage so it is send just once
-                    _sessionsManager.DispatchApplicationMessage(this, willMessage);
+                    await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage);
                }
            }
        }
@@ -133,12 +134,12 @@ namespace MQTTnet.Core.Server
            catch (MqttCommunicationException exception)
            {
                _logger.LogWarning(new EventId(), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
-                Stop();
+                await StopAsync();
            }
            catch (Exception exception)
            {
                _logger.LogError(new EventId(), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
-                Stop();
+                await StopAsync();
            }
        }

@@ -182,14 +183,11 @@ namespace MQTTnet.Core.Server

            if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
            {
-                Stop();
-                return Task.FromResult(0);
+                return StopAsync();
            }

            _logger.LogWarning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
-            Stop();
-
-            return Task.FromResult(0);
+            return StopAsync();            
        }

        private async Task HandleIncomingSubscribePacketAsync(IMqttCommunicationAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
@@ -202,13 +200,13 @@ namespace MQTTnet.Core.Server
            if (subscribeResult.CloseConnection)
            {
                await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false);
-                Stop();
+                await StopAsync();
            }
        }

        private async Task EnqueueSubscribedRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
        {
-            var retainedMessages = await _clientRetainedMessageManager.GetSubscribedMessagesAsync(subscribePacket).ConfigureAwait(false);
+            var retainedMessages = await _sessionsManager.GetRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
            foreach (var publishPacket in retainedMessages)
            {
                EnqueuePublishPacket(publishPacket.ToPublishPacket());
@@ -219,29 +217,16 @@ namespace MQTTnet.Core.Server
        {
            var applicationMessage = publishPacket.ToApplicationMessage();

-            var interceptorContext = new MqttApplicationMessageInterceptorContext
-            {
-                ApplicationMessage = applicationMessage
-            };
-
-            _options.ApplicationMessageInterceptor?.Invoke(interceptorContext);
-            applicationMessage = interceptorContext.ApplicationMessage;
-
-            if (applicationMessage.Retain)
-            {
-                await _clientRetainedMessageManager.HandleMessageAsync(ClientId, applicationMessage).ConfigureAwait(false);
-            }
-
            switch (applicationMessage.QualityOfServiceLevel)
            {
                case MqttQualityOfServiceLevel.AtMostOnce:
                    {
-                        _sessionsManager.DispatchApplicationMessage(this, applicationMessage);
+                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
                        return;
                    }
                case MqttQualityOfServiceLevel.AtLeastOnce:
                    {
-                        _sessionsManager.DispatchApplicationMessage(this, applicationMessage);
+                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);

                        await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
                            new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
@@ -256,7 +241,7 @@ namespace MQTTnet.Core.Server
                            _unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
                        }

-                        _sessionsManager.DispatchApplicationMessage(this, applicationMessage);
+                        await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);

                        await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
                            new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false);
@@ -16,19 +16,24 @@ namespace MQTTnet.Core.Server
 {
    public sealed class MqttClientSessionsManager
    {
-        private readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>();
+        private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
+        private readonly SemaphoreSlim _sessionsSemaphore = new SemaphoreSlim(1, 1);
+
+        private readonly MqttServerOptions _options;
        private readonly ILogger<MqttClientSessionsManager> _logger;
        private readonly IMqttClientSesssionFactory _clientSesssionFactory;
-        private readonly MqttServerOptions _options;
+        private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;

        public MqttClientSessionsManager(
-            IOptions<MqttServerOptions> options, 
+            IOptions<MqttServerOptions> options,
            ILogger<MqttClientSessionsManager> logger,
-            IMqttClientSesssionFactory clientSesssionFactory)
+            IMqttClientSesssionFactory clientSesssionFactory,
+            IMqttClientRetainedMessageManager clientRetainedMessageManager)
        {
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
            _options = options.Value ?? throw new ArgumentNullException(nameof(options));
            _clientSesssionFactory = clientSesssionFactory ?? throw new ArgumentNullException(nameof(clientSesssionFactory));
+            _clientRetainedMessageManager = clientRetainedMessageManager ?? throw new ArgumentNullException(nameof(clientRetainedMessageManager));
        }

        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
@@ -61,7 +66,7 @@ namespace MQTTnet.Core.Server
                    return;
                }

-                var clientSession = GetOrCreateClientSession(connectPacket);
+                var clientSession = await GetOrCreateClientSessionAsync(connectPacket);

                await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
                {
@@ -92,7 +97,7 @@ namespace MQTTnet.Core.Server
                }
                catch (Exception)
                {
-                    //ignored
+                    // ignored
                }

                ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(new ConnectedMqttClient
@@ -103,30 +108,58 @@ namespace MQTTnet.Core.Server
            }
        }

-        public void Clear()
+        public async Task StopAsync()
        {
-            lock (_clientSessions)
+            await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
+            try
            {
-                _clientSessions.Clear();
+                foreach (var session in _sessions)
+                {
+                    await session.Value.StopAsync();
+                }
+
+                _sessions.Clear();
+            }
+            finally
+            {
+                _sessionsSemaphore.Release();
            }
        }

-        public IList<ConnectedMqttClient> GetConnectedClients()
+        public async Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
        {
-            lock (_clientSessions)
+            await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
+            try
            {
-                return _clientSessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
+                return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient
                {
                    ClientId = s.Value.ClientId,
                    ProtocolVersion = s.Value.ProtocolVersion ?? MqttProtocolVersion.V311
                }).ToList();
            }
+            finally
+            {
+                _sessionsSemaphore.Release();
+            }
        }

-        public void DispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        public async Task DispatchApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
        {
            try
            {
+                var interceptorContext = new MqttApplicationMessageInterceptorContext
+                {
+                    ApplicationMessage = applicationMessage
+                };
+
+                _options.ApplicationMessageInterceptor?.Invoke(interceptorContext);
+                applicationMessage = interceptorContext.ApplicationMessage;
+
+                if (applicationMessage.Retain)
+                {
+                    await _clientRetainedMessageManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
+                }
+
                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, applicationMessage);
                ApplicationMessageReceived?.Invoke(this, eventArgs);
            }
@@ -135,15 +168,20 @@ namespace MQTTnet.Core.Server
                _logger.LogError(new EventId(), exception, "Error while processing application message");
            }

-            lock (_clientSessions)
+            lock (_sessions)
            {
-                foreach (var clientSession in _clientSessions.Values.ToList())
+                foreach (var clientSession in _sessions.Values.ToList())
                {
                    clientSession.EnqueuePublishPacket(applicationMessage.ToPublishPacket());
                }
            }
        }

+        public Task<List<MqttApplicationMessage>> GetRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
+        {
+            return _clientRetainedMessageManager.GetSubscribedMessagesAsync(subscribePacket);
+        }
+
        private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
        {
            if (_options.ConnectionValidator != null)
@@ -154,17 +192,18 @@ namespace MQTTnet.Core.Server
            return MqttConnectReturnCode.ConnectionAccepted;
        }

-        private GetOrCreateClientSessionResult GetOrCreateClientSession(MqttConnectPacket connectPacket)
+        private async Task<GetOrCreateClientSessionResult> GetOrCreateClientSessionAsync(MqttConnectPacket connectPacket)
        {
-            lock (_clientSessions)
+            await _sessionsSemaphore.WaitAsync().ConfigureAwait(false);
+            try
            {
-                var isSessionPresent = _clientSessions.TryGetValue(connectPacket.ClientId, out var clientSession);
+                var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession);
                if (isSessionPresent)
                {
                    if (connectPacket.CleanSession)
                    {
-                        _clientSessions.Remove(connectPacket.ClientId);
-                        clientSession.Stop();
+                        _sessions.Remove(connectPacket.ClientId);
+                        await clientSession.StopAsync();
                        clientSession = null;

                        _logger.LogTrace("Stopped existing session of client '{0}'.", connectPacket.ClientId);
@@ -181,13 +220,17 @@ namespace MQTTnet.Core.Server
                    isExistingSession = false;

                    clientSession = _clientSesssionFactory.CreateClientSession(connectPacket.ClientId, this);
-                    _clientSessions[connectPacket.ClientId] = clientSession;
+                    _sessions[connectPacket.ClientId] = clientSession;

                    _logger.LogTrace("Created a new session for client '{0}'.", connectPacket.ClientId);
                }

                return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
            }
+            finally
+            {
+                _sessionsSemaphore.Release();
+            }
        }
    }
 }
@@ -20,12 +20,11 @@ namespace MQTTnet.Core.Server
        private CancellationTokenSource _cancellationTokenSource;

        public MqttServer(
-            IOptions<MqttServerOptions> options, 
+            IOptions<MqttServerOptions> options,
            IEnumerable<IMqttServerAdapter> adapters,
            ILogger<MqttServer> logger,
            MqttClientSessionsManager clientSessionsManager,
-            IMqttClientRetainedMessageManager clientRetainedMessageManager
-            )
+            IMqttClientRetainedMessageManager clientRetainedMessageManager)
        {
            _options = options.Value ?? throw new ArgumentNullException(nameof(options));
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -35,7 +34,7 @@ namespace MQTTnet.Core.Server
            if (adapters == null)
            {
                throw new ArgumentNullException(nameof(adapters));
-            }            
+            }

            _adapters = adapters.ToList();

@@ -44,9 +43,9 @@ namespace MQTTnet.Core.Server
            _clientSessionsManager.ClientDisconnected += OnClientDisconnected;
        }

-        public IList<ConnectedMqttClient> GetConnectedClients()
+        public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
        {
-            return _clientSessionsManager.GetConnectedClients();
+            return _clientSessionsManager.GetConnectedClientsAsync();
        }

        public event EventHandler<MqttServerStartedEventArgs> Started;
@@ -54,7 +53,7 @@ namespace MQTTnet.Core.Server
        public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;

-        public void Publish(IEnumerable<MqttApplicationMessage> applicationMessages)
+        public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
        {
            if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));

@@ -65,23 +64,10 @@ namespace MQTTnet.Core.Server

            foreach (var applicationMessage in applicationMessages)
            {
-                var interceptorContext = new MqttApplicationMessageInterceptorContext
-                {
-                    ApplicationMessage = applicationMessage
-                };
-
-                _options.ApplicationMessageInterceptor?.Invoke(interceptorContext);
-                
-                _clientSessionsManager.DispatchApplicationMessage(null, interceptorContext.ApplicationMessage);
+                await _clientSessionsManager.DispatchApplicationMessageAsync(null, applicationMessage);
            }
        }

-        public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
-        {
-            Publish(applicationMessages);
-            return Task.FromResult(0);
-        }
-
        public async Task StartAsync()
        {
            if (_cancellationTokenSource != null) throw new InvalidOperationException("The MQTT server is already started.");
@@ -113,7 +99,7 @@ namespace MQTTnet.Core.Server
                await adapter.StopAsync();
            }

-            _clientSessionsManager.Clear();
+            await _clientSessionsManager.StopAsync();

            _logger.LogInformation("Stopped.");
        }