Stop existing sessions before takeover from another client.

Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs

@@ -36,8 +36,8 @@ namespace MQTTnet.Server
             {
                 return;
             }
-
-            Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken);
+            
+            Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
         public void Pause()

Source/MQTTnet/Server/MqttClientSession.cs

@@ -9,6 +9,7 @@ using MQTTnet.Exceptions;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Serializer;
 
 namespace MQTTnet.Server
 {
@@ -17,6 +18,7 @@ namespace MQTTnet.Server
         private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
 
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
+        private readonly MqttServerEventDispatcher _eventDispatcher;
         private readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
         private readonly MqttClientPendingPacketsQueue _pendingPacketsQueue;
         private readonly MqttClientSubscriptionsManager _subscriptionsManager;
@@ -28,45 +30,44 @@ namespace MQTTnet.Server
         private CancellationTokenSource _cancellationTokenSource;
         private MqttApplicationMessage _willMessage;
         private bool _wasCleanDisconnect;
-        private IMqttChannelAdapter _adapter;
         private Task _workerTask;
         private IDisposable _cleanupHandle;
 
+        private string _adapterEndpoint;
+        private MqttProtocolVersion? _adapterProtocolVersion;
+        
         public MqttClientSession(
             string clientId,
             IMqttServerOptions options,
             MqttClientSessionsManager sessionsManager,
             MqttRetainedMessagesManager retainedMessagesManager,
+            MqttServerEventDispatcher eventDispatcher,
             IMqttNetChildLogger logger)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            _sessionsManager = sessionsManager;
+            _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
+            _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
 
             ClientId = clientId;
 
             _logger = logger.CreateChildLogger(nameof(MqttClientSession));
 
             _keepAliveMonitor = new MqttClientKeepAliveMonitor(this, _logger);
-            _subscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server);
+            _subscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, eventDispatcher);
             _pendingPacketsQueue = new MqttClientPendingPacketsQueue(_options, this, _logger);
         }
 
         public string ClientId { get; }
 
-        public void ResumeSession()
-        {
-            _keepAliveMonitor.Reset();
-        }
-
         public void FillStatus(MqttClientSessionStatus status)
         {
             status.ClientId = ClientId;
-            status.IsConnected = _adapter != null;
-            status.Endpoint = _adapter?.Endpoint;
-            status.ProtocolVersion = _adapter?.PacketSerializer?.ProtocolVersion;
+            status.IsConnected = _cancellationTokenSource != null;
+            status.Endpoint = _adapterEndpoint;
+            status.ProtocolVersion = _adapterProtocolVersion;
             status.PendingApplicationMessagesCount = _pendingPacketsQueue.Count;
             status.LastPacketReceived = _keepAliveMonitor.LastPacketReceived;
             status.LastNonKeepAlivePacketReceived = _keepAliveMonitor.LastNonKeepAlivePacketReceived;
@@ -85,7 +86,10 @@ namespace MQTTnet.Server
 
             try
             {
-                _adapter = adapter;
+                if (_cancellationTokenSource != null)
+                {
+                    Stop(MqttClientDisconnectType.Clean, true);
+                }
 
                 adapter.ReadingPacketStarted += OnAdapterReadingPacketStarted;
                 adapter.ReadingPacketCompleted += OnAdapterReadingPacketCompleted;
@@ -94,7 +98,7 @@ namespace MQTTnet.Server
 
                 //workaround for https://github.com/dotnet/corefx/issues/24430
 #pragma warning disable 4014
-                _cleanupHandle = _cancellationTokenSource.Token.Register(() => CleanupAsync());
+                _cleanupHandle = _cancellationTokenSource.Token.Register(() => TryDisposeAdapterAsync(adapter));
 #pragma warning restore 4014
                 //end workaround
 
@@ -104,6 +108,9 @@ namespace MQTTnet.Server
                 _pendingPacketsQueue.Start(adapter, _cancellationTokenSource.Token);
                 _keepAliveMonitor.Start(connectPacket.KeepAlivePeriod, _cancellationTokenSource.Token);
 
+                _adapterEndpoint = adapter.Endpoint;
+                _adapterProtocolVersion = adapter.PacketSerializer.ProtocolVersion;
+
                 while (!_cancellationTokenSource.IsCancellationRequested)
                 {
                     var packet = await adapter.ReceivePacketAsync(TimeSpan.Zero, _cancellationTokenSource.Token).ConfigureAwait(false);
@@ -139,30 +146,29 @@ namespace MQTTnet.Server
             }
             finally
             {
-                await CleanupAsync().ConfigureAwait(false);
+                _adapterEndpoint = null;
+                _adapterProtocolVersion = null;
+
+                await TryDisposeAdapterAsync(adapter).ConfigureAwait(false);
 
                 _cleanupHandle?.Dispose();
                 _cleanupHandle = null;
                 
-                _adapter = null;
                 _cancellationTokenSource?.Cancel(false);
                 _cancellationTokenSource?.Dispose();
                 _cancellationTokenSource = null;
             }
         }
 
-        private async Task CleanupAsync()
+        private async Task TryDisposeAdapterAsync(IMqttChannelAdapter adapter)
         {
-            var adapter = _adapter;
+            if (adapter == null)
+            {
+                return;
+            }
+
             try
             {
-                if (adapter == null)
-                {
-                    return;
-                }
-
-                _adapter = null;
-
                 adapter.ReadingPacketStarted -= OnAdapterReadingPacketStarted;
                 adapter.ReadingPacketCompleted -= OnAdapterReadingPacketCompleted;
 
@@ -174,7 +180,13 @@ namespace MQTTnet.Server
             }
             finally
             {
-                adapter?.Dispose();
+                try
+                {
+                    adapter.Dispose();
+                }
+                catch
+                {
+                }
             }
         }
 
@@ -193,10 +205,10 @@ namespace MQTTnet.Server
                     return;
                 }
 
-                _wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
-
                 _cancellationTokenSource?.Cancel(false);
 
+                _wasCleanDisconnect = type == MqttClientDisconnectType.Clean;
+
                 if (_willMessage != null && !_wasCleanDisconnect)
                 {
                     _sessionsManager.EnqueueApplicationMessage(this, _willMessage.ToPublishPacket());
@@ -211,9 +223,8 @@ namespace MQTTnet.Server
             }
             finally
             {
-                _logger.Info("Client '{0}': Session stopped.", ClientId);
-
-                _sessionsManager.Server.OnClientDisconnected(ClientId, _wasCleanDisconnect);
+                _logger.Info("Client '{0}': Disconnected (clean={1}).", ClientId, _wasCleanDisconnect);
+                _eventDispatcher.OnClientDisconnected(ClientId, _wasCleanDisconnect);
             }
         }
 

Source/MQTTnet/Server/MqttClientSessionsManager.cs

@@ -23,25 +23,29 @@ namespace MQTTnet.Server
         private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
 
         private readonly CancellationToken _cancellationToken;
+        private readonly MqttServerEventDispatcher _eventDispatcher;
 
         private readonly MqttRetainedMessagesManager _retainedMessagesManager;
         private readonly IMqttServerOptions _options;
         private readonly IMqttNetChildLogger _logger;
 
-        public MqttClientSessionsManager(IMqttServerOptions options, MqttServer server, MqttRetainedMessagesManager retainedMessagesManager, CancellationToken cancellationToken, IMqttNetChildLogger logger)
+        public MqttClientSessionsManager(
+            IMqttServerOptions options, 
+            MqttRetainedMessagesManager retainedMessagesManager, 
+            CancellationToken cancellationToken,
+            MqttServerEventDispatcher eventDispatcher,
+            IMqttNetChildLogger logger)
         {
-            if (logger == null) throw new ArgumentNullException(nameof(logger));
+            _cancellationToken = cancellationToken;
 
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttClientSessionsManager));
 
-            _cancellationToken = cancellationToken;
+            _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            Server = server ?? throw new ArgumentNullException(nameof(server));
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
-        public MqttServer Server { get; }
-
         public void Start()
         {
             Task.Factory.StartNew(() => TryProcessQueuedApplicationMessages(_cancellationToken), _cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
@@ -176,7 +180,7 @@ namespace MQTTnet.Server
                     applicationMessage = interceptorContext.ApplicationMessage;
                 }
 
-                Server.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
+                _eventDispatcher.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
 
                 if (applicationMessage.Retain)
                 {
@@ -247,8 +251,6 @@ namespace MQTTnet.Server
 
                 var result = PrepareClientSession(connectPacket);
 
-                Server.OnClientConnected(clientId);
-
                 await clientAdapter.SendPacketAsync(
                     new MqttConnAckPacket
                     {
@@ -257,6 +259,9 @@ namespace MQTTnet.Server
                     },
                     cancellationToken).ConfigureAwait(false);
 
+                _logger.Info("Client '{0}': Connected.", clientId);
+                _eventDispatcher.OnClientConnected(clientId);
+
                 await result.Session.RunAsync(connectPacket, clientAdapter).ConfigureAwait(false);
             }
             catch (OperationCanceledException)
@@ -312,8 +317,6 @@ namespace MQTTnet.Server
                     }
                     else
                     {
-                        clientSession.ResumeSession();
-
                         _logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
                     }
                 }
@@ -323,7 +326,7 @@ namespace MQTTnet.Server
                 {
                     isExistingSession = false;
 
-                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _retainedMessagesManager, _logger);
+                    clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _retainedMessagesManager, _eventDispatcher, _logger);
                     _sessions[connectPacket.ClientId] = clientSession;
 
                     _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);

Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs

@@ -10,14 +10,14 @@ namespace MQTTnet.Server
     {
         private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         private readonly IMqttServerOptions _options;
-        private readonly MqttServer _server;
+        private readonly MqttServerEventDispatcher _eventDispatcher;
         private readonly string _clientId;
 
-        public MqttClientSubscriptionsManager(string clientId, IMqttServerOptions options, MqttServer server)
+        public MqttClientSubscriptionsManager(string clientId, IMqttServerOptions options, MqttServerEventDispatcher eventDispatcher)
         {
             _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
             _options = options ?? throw new ArgumentNullException(nameof(options));
-            _server = server;
+            _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
         }
 
         public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket)
@@ -58,7 +58,7 @@ namespace MQTTnet.Server
                         _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
                     }
 
-                    _server.OnClientSubscribedTopic(_clientId, topicFilter);
+                    _eventDispatcher.OnClientSubscribedTopic(_clientId, topicFilter);
                 }
             }
 
@@ -75,7 +75,7 @@ namespace MQTTnet.Server
                 {
                     _subscriptions.Remove(topicFilter);
 
-                    _server.OnClientUnsubscribedTopic(_clientId, topicFilter);
+                    _eventDispatcher.OnClientUnsubscribedTopic(_clientId, topicFilter);
                 }
             }
 

Source/MQTTnet/Server/MqttServer.cs

@@ -11,6 +11,7 @@ namespace MQTTnet.Server
 {
     public class MqttServer : IMqttServer
     {
+        private readonly MqttServerEventDispatcher _eventDispatcher = new MqttServerEventDispatcher();
         private readonly ICollection<IMqttServerAdapter> _adapters;
         private readonly IMqttNetChildLogger _logger;
 
@@ -21,10 +22,16 @@ namespace MQTTnet.Server
         public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetChildLogger logger)
         {
             if (adapters == null) throw new ArgumentNullException(nameof(adapters));
+            _adapters = adapters.ToList();
+
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.CreateChildLogger(nameof(MqttServer));
 
-            _adapters = adapters.ToList();
+            _eventDispatcher.ClientConnected += (s, e) => ClientConnected?.Invoke(s, e);
+            _eventDispatcher.ClientDisconnected += (s, e) => ClientDisconnected?.Invoke(s, e);
+            _eventDispatcher.ClientSubscribedTopic += (s, e) => ClientSubscribedTopic?.Invoke(s, e);
+            _eventDispatcher.ClientUnsubscribedTopic += (s, e) => ClientUnsubscribedTopic?.Invoke(s, e);
+            _eventDispatcher.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
         }
 
         public event EventHandler Started;
@@ -92,7 +99,7 @@ namespace MQTTnet.Server
             _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
             await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
 
-            _clientSessionsManager = new MqttClientSessionsManager(Options, this, _retainedMessagesManager, _cancellationTokenSource.Token, _logger);
+            _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _cancellationTokenSource.Token, _eventDispatcher, _logger);
             _clientSessionsManager.Start();
 
             foreach (var adapter in _adapters)
@@ -144,33 +151,6 @@ namespace MQTTnet.Server
             return _retainedMessagesManager?.ClearMessagesAsync();
         }
 
-        internal void OnClientConnected(string clientId)
-        {
-            _logger.Info("Client '{0}': Connected.", clientId);
-            ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientId));
-        }
-
-        internal void OnClientDisconnected(string clientId, bool wasCleanDisconnect)
-        {
-            _logger.Info("Client '{0}': Disconnected (clean={1}).", clientId, wasCleanDisconnect);
-            ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientId, wasCleanDisconnect));
-        }
-
-        internal void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
-        {
-            ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
-        }
-
-        internal void OnClientUnsubscribedTopic(string clientId, string topicFilter)
-        {
-            ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
-        }
-
-        internal void OnApplicationMessageReceived(string clientId, MqttApplicationMessage applicationMessage)
-        {
-            ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(clientId, applicationMessage));
-        }
-
         private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
         {
             eventArgs.SessionTask = _clientSessionsManager.StartSession(eventArgs.Client);

Source/MQTTnet/Server/MqttServerEventDispatcher.cs

@@ -0,0 +1,42 @@
+using System;
+
+namespace MQTTnet.Server
+{
+    public class MqttServerEventDispatcher
+    {
+        public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
+
+        public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
+
+        public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
+
+        public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
+
+        public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
+
+        public void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
+        {
+            ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
+        }
+
+        public void OnClientUnsubscribedTopic(string clientId, string topicFilter)
+        {
+            ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
+        }
+
+        public void OnClientDisconnected(string clientId, bool wasCleanDisconnect)
+        {
+            ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientId, wasCleanDisconnect));
+        }
+
+        public void OnApplicationMessageReceived(string senderClientId, MqttApplicationMessage applicationMessage)
+        {
+            ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage));
+        }
+
+        public void OnClientConnected(string clientId)
+        {
+            ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientId));
+        }
+    }
+}

Tests/MQTTnet.Core.Tests/MqttServerTests.cs

@@ -228,6 +228,39 @@ namespace MQTTnet.Core.Tests
 
             Assert.AreEqual(2000, receivedMessagesCount);
         }
+        
+        [TestMethod]
+        public async Task MqttServer_SessionTakeover()
+        {
+            var server = new MqttFactory().CreateMqttServer();
+            try
+            {
+                await server.StartAsync(new MqttServerOptions());
+
+                var client1 = new MqttFactory().CreateMqttClient();
+                var client2 = new MqttFactory().CreateMqttClient();
+
+                var options = new MqttClientOptionsBuilder()
+                    .WithTcpServer("localhost")
+                    .WithCleanSession(false)
+                    .WithClientId("a").Build();
+
+                await client1.ConnectAsync(options);
+
+                await Task.Delay(500);
+
+                await client2.ConnectAsync(options);
+
+                await Task.Delay(500);
+
+                Assert.IsFalse(client1.IsConnected);
+                Assert.IsTrue(client2.IsConnected);
+            }
+            finally
+            {
+                await server.StopAsync();
+            }
+        }
 
         private static async Task Publish(IMqttClient c1, MqttApplicationMessage message)
         {

Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs

@@ -1,6 +1,4 @@
 using Microsoft.VisualStudio.TestTools.UnitTesting;
-using MQTTnet.Adapter;
-using MQTTnet.Diagnostics;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
@@ -13,7 +11,7 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeSingleSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServer(new IMqttServerAdapter[0], new MqttNetLogger().CreateChildLogger("")));
+            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServerEventDispatcher());
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
@@ -28,7 +26,7 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeDifferentQoSSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServer(new IMqttServerAdapter[0], new MqttNetLogger().CreateChildLogger("")));
+            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServerEventDispatcher());
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilter("A/B/C", MqttQualityOfServiceLevel.AtMostOnce));
@@ -43,7 +41,7 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeTwoTimesSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServer(new IMqttServerAdapter[0], new MqttNetLogger().CreateChildLogger("")));
+            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServerEventDispatcher());
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce));
@@ -59,7 +57,7 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeSingleNoSuccess()
         {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServer(new IMqttServerAdapter[0], new MqttNetLogger().CreateChildLogger("")));
+            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServerEventDispatcher());
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());
@@ -72,7 +70,7 @@ namespace MQTTnet.Core.Tests
         [TestMethod]
         public void MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle()
         {
-            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServer(new IMqttServerAdapter[0], new MqttNetLogger().CreateChildLogger("")));
+            var sm = new MqttClientSubscriptionsManager("", new MqttServerOptions(), new MqttServerEventDispatcher());
 
             var sp = new MqttSubscribePacket();
             sp.TopicFilters.Add(new TopicFilterBuilder().WithTopic("A/B/C").Build());