1883 in a weak network environment the mqtt broker is unable to detec…

…t disconnections (#1891)

* Add new setting

* Align keep alive handling with RFC

Only completely sent packets are treated as keep alive

* Update ReleaseNotes.md

.github/workflows/ReleaseNotes.md

@@ -1 +1,2 @@
 * [Server] Added new events for delivered and dropped messages (#1866, thanks to @kallayj).
+* [Server] The server will no longer treat a client which is receiving a large payload as alive. The packet must be received completely within the keep alive boundaries (BREAKING CHANGE!, #1883).

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -83,8 +83,6 @@ public string Endpoint
             }
         }
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection
         {
             get
@@ -164,11 +162,6 @@ public async Task<MqttPacket> ReceivePacketAsync(CancellationToken cancellationT
                                 BytesReceived += received;
                                 return packet;
                             }
-                            else
-                            {
-                                // we did receive something but the message is not yet complete
-                                IsReadingPacket = true;
-                            }
                         }
                         else if (readResult.IsCompleted)
                         {
@@ -189,12 +182,9 @@ public async Task<MqttPacket> ReceivePacketAsync(CancellationToken cancellationT
                 // completing the channel makes sure that there is no more data read after a protocol error
                 _input?.Complete(exception);
                 _output?.Complete(exception);
+
                 throw;
             }
-            finally
-            {
-                IsReadingPacket = false;
-            }
 
             cancellationToken.ThrowIfCancellationRequested();
             return null;

Source/MQTTnet/Adapter/IMqttChannelAdapter.cs

@@ -25,8 +25,6 @@ public interface IMqttChannelAdapter : IDisposable
 
         long BytesReceived { get; }
 
-        bool IsReadingPacket { get; }
-
         Task ConnectAsync(CancellationToken cancellationToken);
 
         Task DisconnectAsync(CancellationToken cancellationToken);

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -55,8 +55,6 @@ public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packe
 
         public string Endpoint => _channel.Endpoint;
 
-        public bool IsReadingPacket { get; private set; }
-
         public bool IsSecureConnection => _channel.IsSecureConnection;
 
         public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
@@ -389,53 +387,45 @@ async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellationToken)
                 return ReceivedMqttPacket.Empty;
             }
 
-            IsReadingPacket = true;
-            try
+            var fixedHeader = readFixedHeaderResult.FixedHeader;
+            if (fixedHeader.RemainingLength == 0)
             {
-                var fixedHeader = readFixedHeaderResult.FixedHeader;
-                if (fixedHeader.RemainingLength == 0)
-                {
-                    return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
-                }
+                return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2);
+            }
 
-                var bodyLength = fixedHeader.RemainingLength;
-                var body = new byte[bodyLength];
+            var bodyLength = fixedHeader.RemainingLength;
+            var body = new byte[bodyLength];
 
-                var bodyOffset = 0;
-                var chunkSize = Math.Min(ReadBufferSize, bodyLength);
+            var bodyOffset = 0;
+            var chunkSize = Math.Min(ReadBufferSize, bodyLength);
 
-                do
+            do
+            {
+                var bytesLeft = body.Length - bodyOffset;
+                if (chunkSize > bytesLeft)
                 {
-                    var bytesLeft = body.Length - bodyOffset;
-                    if (chunkSize > bytesLeft)
-                    {
-                        chunkSize = bytesLeft;
-                    }
+                    chunkSize = bytesLeft;
+                }
 
-                    var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
+                var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false);
 
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        return ReceivedMqttPacket.Empty;
-                    }
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    return ReceivedMqttPacket.Empty;
+                }
 
-                    if (readBytes == 0)
-                    {
-                        return ReceivedMqttPacket.Empty;
-                    }
+                if (readBytes == 0)
+                {
+                    return ReceivedMqttPacket.Empty;
+                }
 
-                    bodyOffset += readBytes;
-                } while (bodyOffset < bodyLength);
+                bodyOffset += readBytes;
+            } while (bodyOffset < bodyLength);
 
-                PacketInspector?.FillReceiveBuffer(body);
+            PacketInspector?.FillReceiveBuffer(body);
 
-                var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
-                return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
-            }
-            finally
-            {
-                IsReadingPacket = false;
-            }
+            var bodySegment = new ArraySegment<byte>(body, 0, bodyLength);
+            return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength);
         }
 
         static bool WrapAndThrowException(Exception exception)

Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs

@@ -107,13 +107,6 @@ void TryProcessClient(MqttClient connection, DateTime now)
                     return;
                 }
 
-                if (connection.ChannelAdapter.IsReadingPacket)
-                {
-                    // The connection is currently reading a (large) packet. So it is obviously 
-                    // doing something and thus "connected".
-                    return;
-                }
-
                 // Values described here: [MQTT-3.1.2-24].
                 // If the client sends 5 sec. the server will allow up to 7.5 seconds.
                 // If the client sends 1 sec. the server will allow up to 1.5 seconds.

Source/MQTTnet/Server/Options/MqttServerKeepAliveOptions.cs

@@ -0,0 +1,21 @@
+// 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;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttServerKeepAliveOptions
+    {
+        /// <summary>
+        ///     When this mode is enabled the MQTT server will not close a connection when the
+        ///     client is currently sending a (large) payload. This may lead to "dead" connections
+        ///     When this mode is disabled the MQTT server will disconnect a client when the keep
+        ///     alive timeout is reached even if the client is currently sending a (large) payload.
+        /// </summary>
+        public bool DisconnectClientWhenReadingPayload { get; set; }
+
+        public TimeSpan MonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
+    }
+}

Source/MQTTnet/Server/Options/MqttServerOptions.cs

@@ -9,12 +9,19 @@ namespace MQTTnet.Server
     public sealed class MqttServerOptions
     {
         public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(100);
-        
+
         public MqttServerTcpEndpointOptions DefaultEndpointOptions { get; } = new MqttServerTcpEndpointOptions();
 
         public bool EnablePersistentSessions { get; set; }
 
-        public TimeSpan KeepAliveMonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
+        [Obsolete("Use KeepAliveOptions instead.")]
+        public TimeSpan KeepAliveMonitorInterval
+        {
+            get => KeepAliveOptions.MonitorInterval;
+            set => KeepAliveOptions.MonitorInterval = value;
+        }
+
+        public MqttServerKeepAliveOptions KeepAliveOptions { get; } = new MqttServerKeepAliveOptions();
 
         public int MaxPendingMessagesPerClient { get; set; } = 250;