From 86c348df58e303fca5661e3654cd5a870b7a9a82 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 2 Dec 2018 21:50:16 +0100 Subject: [PATCH] Stop existing sessions before takeover from another client. --- .../Server/MqttClientKeepAliveMonitor.cs | 4 +- Source/MQTTnet/Server/MqttClientSession.cs | 71 +++++++++++-------- .../Server/MqttClientSessionsManager.cs | 27 +++---- .../Server/MqttClientSubscriptionsManager.cs | 10 +-- Source/MQTTnet/Server/MqttServer.cs | 38 +++------- .../Server/MqttServerEventDispatcher.cs | 42 +++++++++++ Tests/MQTTnet.Core.Tests/MqttServerTests.cs | 33 +++++++++ .../MqttSubscriptionsManagerTests.cs | 12 ++-- 8 files changed, 152 insertions(+), 85 deletions(-) create mode 100644 Source/MQTTnet/Server/MqttServerEventDispatcher.cs diff --git a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs index 5abc1640..fbe1f87f 100644 --- a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs +++ b/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() diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 95d6d072..9e402beb 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/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); } } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index b6cea7fa..7b7e3480 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -23,25 +23,29 @@ namespace MQTTnet.Server private readonly Dictionary _sessions = new Dictionary(); 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); diff --git a/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs index 83ac0339..38fad1f4 100644 --- a/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSubscriptionsManager.cs @@ -10,14 +10,14 @@ namespace MQTTnet.Server { private readonly Dictionary _subscriptions = new Dictionary(); 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); } } diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index a960ab00..737d7560 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/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 _adapters; private readonly IMqttNetChildLogger _logger; @@ -21,10 +22,16 @@ namespace MQTTnet.Server public MqttServer(IEnumerable 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); diff --git a/Source/MQTTnet/Server/MqttServerEventDispatcher.cs b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs new file mode 100644 index 00000000..8fd5652d --- /dev/null +++ b/Source/MQTTnet/Server/MqttServerEventDispatcher.cs @@ -0,0 +1,42 @@ +using System; + +namespace MQTTnet.Server +{ + public class MqttServerEventDispatcher + { + public event EventHandler ClientSubscribedTopic; + + public event EventHandler ClientUnsubscribedTopic; + + public event EventHandler ClientConnected; + + public event EventHandler ClientDisconnected; + + public event EventHandler 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)); + } + } +} diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index b86c7b5a..39352ec3 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/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) { diff --git a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs b/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs index 268e5fee..2bdd15f6 100644 --- a/Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs +++ b/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());