From ede8eb86d78661cb4b6e022739402b6804db8e6d Mon Sep 17 00:00:00 2001 From: Jeff Scherrer Date: Thu, 3 Sep 2026 16:17:26 -0700 Subject: [PATCH] Honor disconnect Will reasons and session lifecycle --- .../Internal/MqttClientSessionsManager.cs | 12 + .../Internal/MqttConnectedClient.cs | 15 +- Source/MQTTnet.Server/Internal/MqttSession.cs | 1 + .../Internal/MqttWillMessagesManager.cs | 321 ++++++++++++++++ Source/MQTTnet.Server/MqttServer.cs | 5 + Source/MQTTnet.Tests/Server/Session_Tests.cs | 6 +- .../Server/WillLifecycle_Tests.cs | 348 ++++++++++++++++++ 7 files changed, 692 insertions(+), 16 deletions(-) create mode 100644 Source/MQTTnet.Server/Internal/MqttWillMessagesManager.cs create mode 100644 Source/MQTTnet.Tests/Server/WillLifecycle_Tests.cs diff --git a/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs index 9ea017349..e23910f7b 100644 --- a/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet.Server/Internal/MqttClientSessionsManager.cs @@ -31,6 +31,7 @@ public sealed class MqttClientSessionsManager : ISubscriptionChangedNotification // See the MqttSubscription object for a detailed explanation. readonly MqttSessionsStorage _sessionsStorage = new(); readonly HashSet _subscriberSessions = []; + readonly MqttWillMessagesManager _willMessages; public MqttClientSessionsManager(MqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, MqttServerEventContainer eventContainer, IMqttNetLogger logger) { @@ -42,8 +43,15 @@ public MqttClientSessionsManager(MqttServerOptions options, MqttRetainedMessages _options = options ?? throw new ArgumentNullException(nameof(options)); _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); + _willMessages = new MqttWillMessagesManager(this, logger); } + internal void EndWillSession(MqttSession session) => _willMessages.SessionEnded(session); + + internal Task StartWillsAsync() => _willMessages.StartAsync(); + + internal Task StopWillsAsync(bool endSessions) => _willMessages.StopAsync(endSessions); + public async Task CloseAllConnections(MqttServerClientDisconnectOptions options) { ArgumentNullException.ThrowIfNull(options); @@ -248,6 +256,8 @@ public async Task DispatchApplicationMessage( public void Dispose() { + CloseAllConnections(new MqttServerClientDisconnectOptions()).GetAwaiter().GetResult(); + _willMessages.DisposeAsync().AsTask().GetAwaiter().GetResult(); _createConnectionSyncRoot.Dispose(); _sessionsManagementLock.EnterWriteLock(); @@ -399,6 +409,7 @@ public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter { if (connectedClient != null) { + _willMessages.Disconnected(connectedClient); if (connectedClient.Id != null) { // in case it is a takeover _clientConnections already contains the new connection @@ -593,6 +604,7 @@ async Task CreateClientConnection( connectedClient = CreateClient(connectPacket, channelAdapter, session); _clients[connectPacket.ClientId] = connectedClient; + _willMessages.Connected(connectedClient); } } finally diff --git a/Source/MQTTnet.Server/Internal/MqttConnectedClient.cs b/Source/MQTTnet.Server/Internal/MqttConnectedClient.cs index 5fd70d48b..cbffdfbbc 100644 --- a/Source/MQTTnet.Server/Internal/MqttConnectedClient.cs +++ b/Source/MQTTnet.Server/Internal/MqttConnectedClient.cs @@ -112,19 +112,6 @@ public async Task RunAsync() _cancellationToken = null; } - var isCleanDisconnect = DisconnectPacket != null; - - if (!IsTakenOver && !isCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent) - { - var willPublishPacket = MqttPublishPacketFactory.Create(Session.LatestConnectPacket); - var willApplicationMessage = MqttApplicationMessageFactory.Create(willPublishPacket); - - _ = _sessionsManager.DispatchApplicationMessage(Id, UserName, Session.Items, willApplicationMessage, CancellationToken.None); - Session.WillMessageSent = true; - - _logger.Info("Client '{0}': Published will message", Id); - } - _logger.Info("Client '{0}': Connection stopped", Id); } @@ -581,4 +568,4 @@ async Task TrySendDisconnectPacket(MqttServerClientDisconnectOptions options) _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode); } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet.Server/Internal/MqttSession.cs b/Source/MQTTnet.Server/Internal/MqttSession.cs index 7e3ab1279..c6c3a2cbd 100644 --- a/Source/MQTTnet.Server/Internal/MqttSession.cs +++ b/Source/MQTTnet.Server/Internal/MqttSession.cs @@ -101,6 +101,7 @@ public Task DequeuePacketAsync(CancellationToken cancellation public void Dispose() { + _clientSessionsManager.EndWillSession(this); _packetBus.Dispose(); _subscriptionsManager.Dispose(); } diff --git a/Source/MQTTnet.Server/Internal/MqttWillMessagesManager.cs b/Source/MQTTnet.Server/Internal/MqttWillMessagesManager.cs new file mode 100644 index 000000000..c2ddcfe36 --- /dev/null +++ b/Source/MQTTnet.Server/Internal/MqttWillMessagesManager.cs @@ -0,0 +1,321 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Collections; +using MQTTnet.Diagnostics.Logger; +using MQTTnet.Formatter; +using MQTTnet.Protocol; +using MqttPublishPacketFactory = MQTTnet.Server.Internal.Formatter.MqttPublishPacketFactory; + +namespace MQTTnet.Server.Internal; + +internal sealed class MqttWillMessagesManager : IAsyncDisposable +{ + readonly MqttNetSourceLogger _logger; + readonly MqttClientSessionsManager _sessionsManager; + readonly Dictionary _pending = new(); + readonly HashSet _scheduled = new(); + readonly List _ready = new(); + readonly object _syncRoot = new(); + readonly AsyncLocal _dispatching = new(); + CancellationTokenSource _cancellation; + Task _worker = Task.CompletedTask; + bool _acceptConnections; + bool _flushOnStop; + int _generation; + + public MqttWillMessagesManager(MqttClientSessionsManager sessionsManager, IMqttNetLogger logger) + { + _sessionsManager = sessionsManager; + _logger = logger.WithSource(nameof(MqttWillMessagesManager)); + } + + public async ValueTask DisposeAsync() + { + await StopAsync(true).ConfigureAwait(false); + } + + public void Connected(MqttConnectedClient client) + { + lock (_syncRoot) + { + if (!_acceptConnections) + { + return; + } + + if (_pending.Remove(client.Session, out var previous)) + { + _scheduled.Remove(previous); + // Detach the old Will before replacing its connection metadata. An overdue + // Will cannot be cancelled merely because the timer has not run yet. + if (previous.Owner != null && !previous.Owner.IsRunning) + { + Arm(previous); + } + + if (!IsNormalDisconnect(previous.Owner) && + (previous.Due <= DateTime.UtcNow || (previous.Owner != null && previous.Delay == 0))) + { + previous.Due = DateTime.UtcNow; + _ready.Add(previous); + } + } + + client.Session.WillMessageSent = false; + if (client.ConnectPacket.WillFlag) + { + _pending[client.Session] = new PendingWill + { + Session = client.Session, + Owner = client, + SenderId = client.Id, + UserName = client.UserName, + Items = client.Session.Items, + Message = MqttApplicationMessageFactory.Create(MqttPublishPacketFactory.Create(client.ConnectPacket)), + Delay = client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500 ? client.ConnectPacket.WillDelayInterval : 0 + }; + } + } + } + + public void Disconnected(MqttConnectedClient client) + { + lock (_syncRoot) + { + if (!_pending.TryGetValue(client.Session, out var pending) || pending.Owner != client) + { + return; + } + + if (IsNormalDisconnect(client)) + { + _pending.Remove(client.Session); + return; + } + + Arm(pending); + pending.Owner = null; + _scheduled.Add(pending); + } + } + + public void SessionEnded(MqttSession session) + { + lock (_syncRoot) + { + if (_pending.Remove(session, out var pending) && !IsNormalDisconnect(pending.Owner)) + { + _scheduled.Remove(pending); + pending.Due = DateTime.UtcNow; + _ready.Add(pending); + } + } + } + + public async Task StartAsync() + { + CancellationToken token; + int generation; + lock (_syncRoot) + { + _acceptConnections = true; + _flushOnStop = false; + _cancellation = new CancellationTokenSource(); + token = _cancellation.Token; + generation = ++_generation; + } + + // Deliver overdue Wills before the adapters begin accepting resumed sessions. + await DispatchReadyAsync(generation).ConfigureAwait(false); + lock (_syncRoot) + { + if (_acceptConnections && _generation == generation) + { + _worker = Task.Run(() => RunAsync(generation, token), token); + } + } + } + + public async Task StopAsync(bool endSessions) + { + CancellationTokenSource cancellation; + Task worker; + int generation; + lock (_syncRoot) + { + _acceptConnections = false; + _flushOnStop |= endSessions; + endSessions = _flushOnStop; + generation = ++_generation; + // StopAsync on a connection does not wait for its receive-loop cleanup. Capture + // every transition now; a late Disconnected call cannot arm the Will twice. + foreach (var pair in _pending.ToList()) + { + var pending = pair.Value; + if (IsNormalDisconnect(pending.Owner)) + { + _pending.Remove(pair.Key); + _scheduled.Remove(pending); + continue; + } + + if (pending.Owner != null) + { + Arm(pending); + pending.Owner = null; + } + + if (endSessions) + { + pending.Due = DateTime.UtcNow; + } + + _scheduled.Add(pending); + } + + cancellation = _cancellation; + worker = _worker; + cancellation?.Cancel(); + } + + try + { + // A publication callback may stop the server. It must not await itself. + if (!_dispatching.Value) + { + await worker.ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + } + + lock (_syncRoot) + { + if (_cancellation == cancellation) + { + _cancellation = null; + } + } + + cancellation?.Dispose(); + if (endSessions && !_dispatching.Value) + { + await DispatchReadyAsync(generation, true).ConfigureAwait(false); + } + } + + static bool IsNormalDisconnect(MqttConnectedClient client) + { + return client?.DisconnectPacket?.ReasonCode == MqttDisconnectReasonCode.NormalDisconnection; + } + + static void Arm(PendingWill pending) + { + // Session expiry is the effective interval owned by the session lifecycle. + // DateTime supports the entire uint interval range without Task.Delay overflow. + pending.Due = DateTime.UtcNow.AddSeconds(Math.Min(pending.Delay, pending.Session.ExpiryInterval)); + } + + async Task DispatchReadyAsync(int generation, bool flush = false) + { + lock (_syncRoot) + { + MoveDueWills(); + } + + while (true) + { + PendingWill pending; + lock (_syncRoot) + { + if (_generation != generation || (!flush && !_acceptConnections)) + { + if (!_acceptConnections && _flushOnStop) + { + // A callback ended the sessions. Let this dispatch loop finish the + // flush after the callback returns instead of recursively invoking it. + generation = _generation; + flush = true; + MoveDueWills(); + } + else + { + return; + } + } + + // Claim one message at a time. If a callback stops the server, the rest + // remain owned by the scheduler for restart or final session disposal. + pending = _ready.FirstOrDefault(p => p.Owner == null || !p.Owner.IsRunning); + if (pending == null) + { + return; + } + + _ready.Remove(pending); + } + + var wasDispatching = _dispatching.Value; + _dispatching.Value = true; + try + { + // Message Expiry starts here, not when the delayed Will was registered. + await _sessionsManager.DispatchApplicationMessage(pending.SenderId, pending.UserName, pending.Items, pending.Message, CancellationToken.None) + .ConfigureAwait(false); + } + catch (Exception exception) + { + _logger.Error(exception, "Error publishing Will for client '{0}'", pending.SenderId); + } + finally + { + _dispatching.Value = wasDispatching; + } + } + } + + async Task RunAsync(int generation, CancellationToken cancellationToken) + { + using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(100)); + try + { + while (await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false)) + { + await DispatchReadyAsync(generation).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + } + } + + void MoveDueWills() + { + var now = DateTime.UtcNow; + foreach (var pending in _scheduled.ToList()) + { + if (pending.Due <= now) + { + _scheduled.Remove(pending); + _pending.Remove(pending.Session); + pending.Session.WillMessageSent = true; + _ready.Add(pending); + } + } + } + + sealed class PendingWill + { + public MqttSession Session; + public MqttConnectedClient Owner; + public string SenderId; + public string UserName; + public IDictionary Items; + public MqttApplicationMessage Message; + public uint Delay; + public DateTime? Due; + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Server/MqttServer.cs b/Source/MQTTnet.Server/MqttServer.cs index 8f1807cff..5cea7e843 100644 --- a/Source/MQTTnet.Server/MqttServer.cs +++ b/Source/MQTTnet.Server/MqttServer.cs @@ -279,6 +279,8 @@ public async Task StartAsync() await _retainedMessagesManager.Start().ConfigureAwait(false); _clientSessionsManager.Start(); + await _clientSessionsManager.StartWillsAsync().ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); _keepAliveMonitor.Start(cancellationToken); foreach (var adapter in _adapters) @@ -314,6 +316,8 @@ public async Task StopAsync(MqttServerStopOptions options) adapter.ClientHandler = null; await adapter.StopAsync().ConfigureAwait(false); } + + await _clientSessionsManager.StopWillsAsync(!_options.EnablePersistentSessions).ConfigureAwait(false); } finally { @@ -368,6 +372,7 @@ protected override void Dispose(bool disposing) if (disposing) { StopAsync(new MqttServerStopOptions()).GetAwaiter().GetResult(); + _clientSessionsManager.StopWillsAsync(true).GetAwaiter().GetResult(); foreach (var adapter in _adapters) { diff --git a/Source/MQTTnet.Tests/Server/Session_Tests.cs b/Source/MQTTnet.Tests/Server/Session_Tests.cs index 98bc4a34f..20517d123 100644 --- a/Source/MQTTnet.Tests/Server/Session_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Session_Tests.cs @@ -309,7 +309,7 @@ public async Task Use_Clean_Session() } [TestMethod] - public async Task Will_Message_Do_Not_Send_On_Takeover() + public async Task Will_Message_Is_Sent_On_Takeover() { using var testEnvironment = CreateTestEnvironment(); var receivedMessagesCount = 0; @@ -336,7 +336,9 @@ public async Task Will_Message_Do_Not_Send_On_Takeover() await Task.Delay(1000); - Assert.AreEqual(0, receivedMessagesCount); + // MQTT 3.1.1 [3.1.2-8] and [3.1.4-2]: takeover closes the old + // connection without receiving its DISCONNECT, so its Will is published. + Assert.AreEqual(1, receivedMessagesCount); } static async Task ConnectAndSubscribe(TestEnvironment testEnvironment, MqttClientOptionsBuilder options, Action onReceive) diff --git a/Source/MQTTnet.Tests/Server/WillLifecycle_Tests.cs b/Source/MQTTnet.Tests/Server/WillLifecycle_Tests.cs new file mode 100644 index 000000000..58fb3ac5a --- /dev/null +++ b/Source/MQTTnet.Tests/Server/WillLifecycle_Tests.cs @@ -0,0 +1,348 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Collections.Concurrent; +using MQTTnet.Formatter; +using MQTTnet.Protocol; +using MQTTnet.Server; +using MQTTnet.Tests.Mockups; + +namespace MQTTnet.Tests.Server; + +[TestClass] +public sealed class WillLifecycle_Tests +{ + [TestMethod] + [DataRow(MqttClientDisconnectOptionsReason.NormalDisconnection, false)] + [DataRow(MqttClientDisconnectOptionsReason.DisconnectWithWillMessage, true)] + [DataRow(MqttClientDisconnectOptionsReason.UnspecifiedError, true)] + public async Task Disconnect_Reason_Controls_Will(MqttClientDisconnectOptionsReason reason, bool expected) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var will = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var count = 0; + server.InterceptingPublishAsync += _ => { Interlocked.Increment(ref count); will.TrySetResult(); return Task.CompletedTask; }; + var sender = await environment.ConnectClient(Options("sender", 0, 30)); + await sender.DisconnectAsync(new MqttClientDisconnectOptionsBuilder().WithReason(reason).Build()); + if (expected) + { + await will.Task.WaitAsync(TimeSpan.FromSeconds(3)); + } + + await Task.Delay(200); + Assert.AreEqual(expected ? 1 : 0, count); + } + + [TestMethod] + public async Task Abnormal_Disconnect_Observes_Will_Delay() + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var will = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + server.InterceptingPublishAsync += _ => { will.TrySetResult(); return Task.CompletedTask; }; + var sender = await environment.ConnectClient(Options("sender", 1, 30)); + sender.Dispose(); + await Task.Delay(200); + Assert.IsFalse(will.Task.IsCompleted); + await will.Task.WaitAsync(TimeSpan.FromSeconds(3)); + } + + [TestMethod] + [DataRow(0U)] + [DataRow(1U)] + public async Task Session_Expiry_Limits_Will_Delay(uint expiry) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var sender = await environment.ConnectClient(Options("sender", uint.MaxValue, expiry)); + await sender.DisconnectAsync(new MqttClientDisconnectOptionsBuilder().WithReason(MqttClientDisconnectOptionsReason.DisconnectWithWillMessage).Build()); + if (expiry > 0) + { + await Task.Delay(200); + Assert.IsEmpty(messages); + } + + await WaitForCount(messages, 1); + } + + [TestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task Reconnect_Resumes_Or_Ends_Pending_Will(bool cleanStart) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var disconnected = Disconnected(server); + var sender = await environment.ConnectClient(Options("sender", 1, 30)); + sender.Dispose(); + await disconnected.WaitAsync(TimeSpan.FromSeconds(3)); + var resumed = await environment.ConnectClient(NoWill("sender", cleanStart)); + await Task.Delay(1300); + Assert.HasCount(cleanStart ? 1 : 0, messages); + await resumed.DisconnectAsync(); + } + + [TestMethod] + [DataRow(0U, false, true)] + [DataRow(1U, false, false)] + [DataRow(1U, true, true)] + public async Task Takeover_Uses_Old_Connection_Will(uint delay, bool cleanStart, bool expectWill) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + await environment.ConnectClient(Options("sender", delay, 30).WithCredentials("old-user")); + var resumed = await environment.ConnectClient(NoWill("sender", cleanStart).WithCredentials("new-user")); + await Task.Delay(1300); + Assert.HasCount(expectWill ? 1 : 0, messages); + if (expectWill) + { + Assert.AreEqual("will/sender", messages.Single().ApplicationMessage.Topic); + Assert.AreEqual("old-user", messages.Single().UserName); + } + + await resumed.DisconnectAsync(); + } + + [TestMethod] + public async Task Reconnect_After_Expiry_Does_Not_Cancel_Will() + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var disconnected = Disconnected(server); + var sender = await environment.ConnectClient(Options("sender", 30, 1)); + sender.Dispose(); + await disconnected.WaitAsync(TimeSpan.FromSeconds(3)); + await Task.Delay(1200); + var resumed = await environment.ConnectClient(NoWill("sender", false)); + await WaitForCount(messages, 1); + await resumed.DisconnectAsync(); + } + + [TestMethod] + [DataRow(0)] + [DataRow(1000)] + public async Task Session_Deletion_And_Timer_Publish_Only_Once(int deletionDelay) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var disconnected = Disconnected(server); + var sender = await environment.ConnectClient(Options("sender", 1, 30)); + sender.Dispose(); + await disconnected.WaitAsync(TimeSpan.FromSeconds(3)); + var session = (await server.GetSessionsAsync()).Single(); + await Task.Delay(deletionDelay); + await session.DeleteAsync(); + await WaitForCount(messages, 1); + await Task.Delay(300); + Assert.HasCount(1, messages); + } + + [TestMethod] + public async Task Disabled_Persistence_Ends_Will_Delay() + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(); + var messages = Capture(server); + var sender = await environment.ConnectClient(Options("sender", uint.MaxValue, 30)); + sender.Dispose(); + await WaitForCount(messages, 1); + } + + [TestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task Stop_Restart_Preserves_Pending_Deadline(bool disconnectFirst) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var disconnected = Disconnected(server); + var sender = await environment.ConnectClient(Options("sender", 1, 30)); + if (disconnectFirst) + { + sender.Dispose(); + await disconnected.WaitAsync(TimeSpan.FromSeconds(3)); + } + + await server.StopAsync(); + await Task.Delay(1200); + Assert.IsEmpty(messages); + await server.StartAsync(); + await WaitForCount(messages, 1); + } + + [TestMethod] + public async Task Rejected_Reconnect_Does_Not_Cancel_Pending_Will() + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var disconnected = Disconnected(server); + var sender = await environment.ConnectClient(Options("sender", 1, 30)); + sender.Dispose(); + await disconnected.WaitAsync(TimeSpan.FromSeconds(3)); + server.ValidatingConnectionAsync += args => { args.ReasonCode = MqttConnectReasonCode.NotAuthorized; return Task.CompletedTask; }; + var rejected = environment.CreateClient(); + var result = await rejected.ConnectAsync(NoWill("sender", false).WithTcpServer("127.0.0.1", environment.ServerPort).Build()); + Assert.AreEqual(MqttClientConnectResultCode.NotAuthorized, result.ResultCode); + await WaitForCount(messages, 1); + } + + [TestMethod] + public async Task Resumed_Connection_Owns_Its_New_Will() + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + await environment.ConnectClient(Options("sender", 1, 30)); + var resumed = await environment.ConnectClient(Options("sender", 0, 30, false).WithWillTopic("will/new")); + await resumed.DisconnectAsync(new MqttClientDisconnectOptionsBuilder().WithReason(MqttClientDisconnectOptionsReason.DisconnectWithWillMessage).Build()); + await WaitForCount(messages, 1); + await Task.Delay(1200); + Assert.HasCount(1, messages); + Assert.AreEqual("will/new", messages.Single().ApplicationMessage.Topic); + } + + [TestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task Will_Callback_Can_Stop_Server_Without_Losing_The_Next_Will(bool duringRestart) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var first = await environment.ConnectClient(Options("first", 1, 30)); + var second = await environment.ConnectClient(Options("second", 1, 30)); + await first.DisconnectAsync(new MqttClientDisconnectOptionsBuilder().WithReason(MqttClientDisconnectOptionsReason.DisconnectWithWillMessage).Build()); + await second.DisconnectAsync(new MqttClientDisconnectOptionsBuilder().WithReason(MqttClientDisconnectOptionsReason.DisconnectWithWillMessage).Build()); + if (duringRestart) + { + await server.StopAsync(); + await Task.Delay(1200); + } + + var stopped = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var stopOnce = 1; + server.InterceptingPublishAsync += async _ => + { + if (Interlocked.Exchange(ref stopOnce, 0) == 1) + { + await server.StopAsync(); + stopped.TrySetResult(); + } + }; + if (duringRestart) + { + await Assert.ThrowsAsync(() => server.StartAsync().WaitAsync(TimeSpan.FromSeconds(3))); + } + + await stopped.Task.WaitAsync(TimeSpan.FromSeconds(3)); + await Task.Delay(200); + Assert.HasCount(1, messages); + await server.StartAsync(); + await WaitForCount(messages, 2); + } + + [TestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task Nonpersistent_Shutdown_Does_Not_Recursively_Invoke_Will_Callbacks(bool externalStop) + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(); + var messages = Capture(server); + var first = await environment.ConnectClient(Options("first", 0, 30)); + await environment.ConnectClient(Options("second", uint.MaxValue, 30)); + var depth = 0; + var maxDepth = 0; + server.InterceptingPublishAsync += async _ => + { + depth++; + maxDepth = Math.Max(maxDepth, depth); + await server.StopAsync(); + depth--; + }; + if (externalStop) + { + await server.StopAsync().WaitAsync(TimeSpan.FromSeconds(3)); + } + else + { + // The first Will stops the server while the second connection is still active. + first.Dispose(); + } + + await WaitForCount(messages, 2); + Assert.AreEqual(1, maxDepth); + } + + [TestMethod] + public async Task Final_Disposal_Publishes_Pending_Will() + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + await environment.ConnectClient(Options("sender", uint.MaxValue, 30)); + server.Dispose(); + Assert.HasCount(1, messages); + } + + [TestMethod] + public async Task Delayed_Will_Retains_Message_Expiry_And_Retain_Flag() + { + using var environment = new TestEnvironment(); + var server = await environment.StartServer(o => o.WithPersistentSessions()); + var messages = Capture(server); + var sender = await environment.ConnectClient(Options("sender", 1, 30).WithWillRetain().WithWillMessageExpiryInterval(10)); + sender.Dispose(); + await WaitForCount(messages, 1); + Assert.AreEqual(10U, messages.Single().ApplicationMessage.MessageExpiryInterval); + Assert.IsTrue(messages.Single().ApplicationMessage.Retain); + Assert.HasCount(1, await server.GetRetainedMessagesAsync()); + } + + static ConcurrentQueue Capture(MqttServer server) + { + var messages = new ConcurrentQueue(); + server.InterceptingPublishAsync += args => { messages.Enqueue(args); return Task.CompletedTask; }; + return messages; + } + + static Task Disconnected(MqttServer server) + { + var disconnected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + server.ClientDisconnectedAsync += _ => { disconnected.TrySetResult(); return Task.CompletedTask; }; + return disconnected.Task; + } + + static async Task WaitForCount(ConcurrentQueue messages, int count) + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + while (messages.Count < count) + { + await Task.Delay(20, timeout.Token); + } + + Assert.HasCount(count, messages); + } + + static MqttClientOptionsBuilder NoWill(string id, bool clean) + { + return new MqttClientOptionsBuilder().WithClientId(id).WithProtocolVersion(MqttProtocolVersion.V500) + .WithCleanSession(clean).WithSessionExpiryInterval(30); + } + + static MqttClientOptionsBuilder Options(string id, uint delay, uint expiry, bool clean = true) + { + return new MqttClientOptionsBuilder().WithClientId(id).WithProtocolVersion(MqttProtocolVersion.V500) + .WithCleanSession(clean).WithSessionExpiryInterval(expiry).WithWillTopic("will/" + id) + .WithWillPayload("synthetic").WithWillDelayInterval(delay); + } +} \ No newline at end of file