Honor disconnect Will reasons and session lifecycle

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<MqttSession> _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<DispatchApplicationMessageResult> 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<MqttConnectedClient> CreateClientConnection(
 
                     connectedClient = CreateClient(connectPacket, channelAdapter, session);
                     _clients[connectPacket.ClientId] = connectedClient;
+                    _willMessages.Connected(connectedClient);
                 }
             }
             finally

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);
         }
     }
-}
+}

Source/MQTTnet.Server/Internal/MqttSession.cs

@@ -101,6 +101,7 @@ public Task<MqttPacketBusItem> DequeuePacketAsync(CancellationToken cancellation
 
     public void Dispose()
     {
+        _clientSessionsManager.EndWillSession(this);
         _packetBus.Dispose();
         _subscriptionsManager.Dispose();
     }