From 4df06980a15d88c376be20f7430b610d1771a90e Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Thu, 7 Dec 2023 17:49:03 +0100 Subject: [PATCH] 1883 in a weak network environment the mqtt broker is unable to detect 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 + .../MqttConnectionContext.cs | 12 +--- Source/MQTTnet/Adapter/IMqttChannelAdapter.cs | 2 - Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 68 ++++++++----------- .../Internal/MqttServerKeepAliveMonitor.cs | 7 -- .../Options/MqttServerKeepAliveOptions.cs | 21 ++++++ .../Server/Options/MqttServerOptions.cs | 11 ++- 7 files changed, 61 insertions(+), 61 deletions(-) create mode 100644 Source/MQTTnet/Server/Options/MqttServerKeepAliveOptions.cs diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 2cd5a5cd..61707d85 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.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). diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 46d91412..8638af6f 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -83,8 +83,6 @@ namespace MQTTnet.AspNetCore } } - public bool IsReadingPacket { get; private set; } - public bool IsSecureConnection { get @@ -164,11 +162,6 @@ namespace MQTTnet.AspNetCore 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 @@ namespace MQTTnet.AspNetCore // 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; diff --git a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs index 88930eef..7cd7fd1d 100644 --- a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs @@ -25,8 +25,6 @@ namespace MQTTnet.Adapter long BytesReceived { get; } - bool IsReadingPacket { get; } - Task ConnectAsync(CancellationToken cancellationToken); Task DisconnectAsync(CancellationToken cancellationToken); diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index a5842c16..9015825b 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -55,8 +55,6 @@ namespace MQTTnet.Adapter public string Endpoint => _channel.Endpoint; - public bool IsReadingPacket { get; private set; } - public bool IsSecureConnection => _channel.IsSecureConnection; public MqttPacketFormatterAdapter PacketFormatterAdapter { get; } @@ -389,53 +387,45 @@ namespace MQTTnet.Adapter 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); + } + + var bodyLength = fixedHeader.RemainingLength; + var body = new byte[bodyLength]; + + var bodyOffset = 0; + var chunkSize = Math.Min(ReadBufferSize, bodyLength); + + do + { + var bytesLeft = body.Length - bodyOffset; + if (chunkSize > bytesLeft) { - return new ReceivedMqttPacket(fixedHeader.Flags, EmptyBuffer.ArraySegment, 2); + chunkSize = bytesLeft; } - var bodyLength = fixedHeader.RemainingLength; - var body = new byte[bodyLength]; + var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false); - var bodyOffset = 0; - var chunkSize = Math.Min(ReadBufferSize, bodyLength); - - do + if (cancellationToken.IsCancellationRequested) { - var bytesLeft = body.Length - bodyOffset; - if (chunkSize > bytesLeft) - { - chunkSize = bytesLeft; - } + return ReceivedMqttPacket.Empty; + } - var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false); + if (readBytes == 0) + { + return ReceivedMqttPacket.Empty; + } - if (cancellationToken.IsCancellationRequested) - { - return ReceivedMqttPacket.Empty; - } + bodyOffset += readBytes; + } while (bodyOffset < bodyLength); - if (readBytes == 0) - { - return ReceivedMqttPacket.Empty; - } + PacketInspector?.FillReceiveBuffer(body); - bodyOffset += readBytes; - } while (bodyOffset < bodyLength); - - PacketInspector?.FillReceiveBuffer(body); - - var bodySegment = new ArraySegment(body, 0, bodyLength); - return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength); - } - finally - { - IsReadingPacket = false; - } + var bodySegment = new ArraySegment(body, 0, bodyLength); + return new ReceivedMqttPacket(fixedHeader.Flags, bodySegment, fixedHeader.TotalLength); } static bool WrapAndThrowException(Exception exception) diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index ebad68d1..441f007b 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -107,13 +107,6 @@ namespace MQTTnet.Server 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. diff --git a/Source/MQTTnet/Server/Options/MqttServerKeepAliveOptions.cs b/Source/MQTTnet/Server/Options/MqttServerKeepAliveOptions.cs new file mode 100644 index 00000000..58f8a673 --- /dev/null +++ b/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 + { + /// + /// 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. + /// + public bool DisconnectClientWhenReadingPayload { get; set; } + + public TimeSpan MonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Options/MqttServerOptions.cs b/Source/MQTTnet/Server/Options/MqttServerOptions.cs index 59e46ab6..ce0c7e1e 100644 --- a/Source/MQTTnet/Server/Options/MqttServerOptions.cs +++ b/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;