From fb2be796737664233171a855ebd10d3aa2e055a3 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 2 Nov 2020 21:45:23 +0100 Subject: [PATCH] Remove not used timeout for receiving. --- .../MqttConnectionContext.cs | 2 +- Source/MQTTnet/Adapter/IMqttChannelAdapter.cs | 2 +- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 23 ++++++------------- Source/MQTTnet/Client/MqttClient.cs | 2 +- .../LowLevelClient/LowLevelMqttClient.cs | 7 +++--- Source/MQTTnet/Server/MqttClientConnection.cs | 2 +- .../Server/MqttClientSessionsManager.cs | 21 +++++++++++++---- .../Server/MqttServerKeepAliveMonitor.cs | 5 +++- .../MqttConnectionContextTest.cs | 6 ++--- .../ChannelAdapterBenchmark.cs | 2 +- .../Mockups/TestEnvironment.cs | 1 + .../Mockups/TestMqttCommunicationAdapter.cs | 2 +- Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs | 2 ++ .../Server_KeepAlive_Tests.cs | 8 ++++--- Tests/MQTTnet.Core.Tests/Session_Tests.cs | 6 ++--- 15 files changed, 49 insertions(+), 42 deletions(-) diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index 67d814e8..4f743f71 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -92,7 +92,7 @@ namespace MQTTnet.AspNetCore return Task.CompletedTask; } - public async Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) + public async Task ReceivePacketAsync(CancellationToken cancellationToken) { var input = Connection.Transport.Input; diff --git a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs index 78c77f1f..73ce3c2e 100644 --- a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Adapter Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken); - Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken); + Task ReceivePacketAsync(CancellationToken cancellationToken); void ResetStatistics(); } diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 20dc952d..b184b0ab 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -155,42 +155,33 @@ namespace MQTTnet.Adapter } } - public async Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) + public async Task ReceivePacketAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); ThrowIfDisposed(); try { - ReceivedMqttPacket receivedMqttPacket; - if (timeout == TimeSpan.Zero) - { - receivedMqttPacket = await ReceiveAsync(cancellationToken).ConfigureAwait(false); - } - else - { - receivedMqttPacket = await MqttTaskTimeout.WaitAsync(ReceiveAsync, timeout, cancellationToken).ConfigureAwait(false); - } - - if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested) + var receivedPacket = await ReceiveAsync(cancellationToken).ConfigureAwait(false); + if (receivedPacket == null || cancellationToken.IsCancellationRequested) { return null; } - Interlocked.Add(ref _bytesSent, receivedMqttPacket.TotalLength); + Interlocked.Add(ref _bytesSent, receivedPacket.TotalLength); if (PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.Unknown) { - PacketFormatterAdapter.DetectProtocolVersion(receivedMqttPacket); + PacketFormatterAdapter.DetectProtocolVersion(receivedPacket); } - var packet = PacketFormatterAdapter.Decode(receivedMqttPacket); + var packet = PacketFormatterAdapter.Decode(receivedPacket); if (packet == null) { throw new MqttProtocolViolationException("Received malformed packet."); } - _logger.Verbose("RX ({0} bytes) <<< {1}", receivedMqttPacket.TotalLength, packet); + _logger.Verbose("RX ({0} bytes) <<< {1}", receivedPacket.TotalLength, packet); return packet; } diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 06ed62c3..6b9bc3f0 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -493,7 +493,7 @@ namespace MQTTnet.Client while (!cancellationToken.IsCancellationRequested) { - var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); + var packet = await _adapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false); if (cancellationToken.IsCancellationRequested) { diff --git a/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs index 7bb58639..9b88ad8b 100644 --- a/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs +++ b/Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs @@ -18,10 +18,9 @@ namespace MQTTnet.LowLevelClient public LowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory, IMqttNetLogger logger) { - if (clientAdapterFactory is null) throw new ArgumentNullException(nameof(clientAdapterFactory)); - if (logger is null) throw new ArgumentNullException(nameof(logger)); + _clientAdapterFactory = clientAdapterFactory ?? throw new ArgumentNullException(nameof(clientAdapterFactory)); - _clientAdapterFactory = clientAdapterFactory; + if (logger is null) throw new ArgumentNullException(nameof(logger)); _logger = logger.CreateScopedLogger(nameof(LowLevelMqttClient)); } @@ -95,7 +94,7 @@ namespace MQTTnet.LowLevelClient try { - return await _adapter.ReceivePacketAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); + return await _adapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false); } catch (Exception) { diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 07c1825e..a1ee8798 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -167,7 +167,7 @@ namespace MQTTnet.Server { Status = MqttClientConnectionStatus.Running; - var packet = await _channelAdapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); + var packet = await _channelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false); if (packet == null) { // The client has closed the connection gracefully. diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 5cad56f3..80c60210 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -12,6 +12,7 @@ using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; +using OperationCanceledException = System.OperationCanceledException; namespace MQTTnet.Server { @@ -59,14 +60,24 @@ namespace MQTTnet.Server MqttConnectPacket connectPacket; try { - var firstPacket = await channelAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, cancellationToken).ConfigureAwait(false); - connectPacket = firstPacket as MqttConnectPacket; - if (connectPacket == null) + using (var timeoutToken = new CancellationTokenSource(_options.DefaultCommunicationTimeout)) { - _logger.Warning(null, "The first packet from client '{0}' was no 'CONNECT' packet [MQTT-3.1.0-1].", channelAdapter.Endpoint); - return; + var firstPacket = await channelAdapter.ReceivePacketAsync(timeoutToken.Token).ConfigureAwait(false); + connectPacket = firstPacket as MqttConnectPacket; + if (connectPacket == null) + { + _logger.Warning(null, + "The first packet from client '{0}' was no 'CONNECT' packet [MQTT-3.1.0-1].", + channelAdapter.Endpoint); + return; + } } } + catch (OperationCanceledException) + { + _logger.Warning(null, "Client '{0}' connected but did not sent a CONNECT packet.", channelAdapter.Endpoint); + return; + } catch (MqttCommunicationTimedOutException) { _logger.Warning(null, "Client '{0}' connected but did not sent a CONNECT packet.", channelAdapter.Endpoint); diff --git a/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs index f90f0dca..7b050b3a 100644 --- a/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/MqttServerKeepAliveMonitor.cs @@ -105,7 +105,10 @@ namespace MQTTnet.Server // Execute the disconnection in background so that the keep alive monitor can continue // with checking other connections. - Task.Run(() => connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout)); + // We do not need to wait for the task so no await is needed. + // Also the internal state of the connection must be swapped to "Finalizing" because the + // next iteration of the keep alive timer happens. + var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout); } catch (Exception exception) { diff --git a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs index a81ffec2..9e4dc569 100644 --- a/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs +++ b/Tests/MQTTnet.AspNetCore.Tests/MqttConnectionContextTest.cs @@ -34,7 +34,7 @@ namespace MQTTnet.AspNetCore.Tests pipe.Receive.Writer.Complete(); - await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None)); + await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); } [TestMethod] @@ -49,10 +49,10 @@ namespace MQTTnet.AspNetCore.Tests await pipe.Receive.Writer.WriteAsync(writer.AddMqttHeader(MqttControlPacketType.Connect, new byte[0])); - await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None)); + await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); // the first exception should complete the pipes so if someone tries to use the connection after that it should throw immidiatly - await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None)); + await Assert.ThrowsExceptionAsync(() => ctx.ReceivePacketAsync(CancellationToken.None)); } // COMMENTED OUT DUE TO DEAD LOCK? OR VERY VERY SLOW PERFORMANCE ON LOCAL DEV MACHINE. TEST WAS STILL RUNNING AFTER SEVERAL MINUTES! diff --git a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs index 47e1c971..20323f40 100644 --- a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs @@ -53,7 +53,7 @@ namespace MQTTnet.Benchmarks for (var i = 0; i < 10000; i++) { - _channelAdapter.ReceivePacketAsync(TimeSpan.Zero, CancellationToken.None).GetAwaiter().GetResult(); + _channelAdapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult(); } _stream.Position = 0; diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index b0e798cd..94f0b612 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -169,6 +169,7 @@ namespace MQTTnet.Tests.Mockups { foreach (var mqttClient in _clients) { + mqttClient.DisconnectAsync().GetAwaiter().GetResult(); mqttClient?.Dispose(); } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs index 833160f2..89a4216b 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapter.cs @@ -51,7 +51,7 @@ namespace MQTTnet.Tests.Mockups return Task.FromResult(0); } - public Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) + public Task ReceivePacketAsync(CancellationToken cancellationToken) { ThrowIfPartnerIsNull(); diff --git a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs index 16405de2..b1774f74 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs @@ -161,6 +161,8 @@ namespace MQTTnet.Tests SpinWait.SpinUntil(() => replyReceived, TimeSpan.FromSeconds(10)); + await Task.Delay(500); + Assert.IsTrue(replyReceived); } } diff --git a/Tests/MQTTnet.Core.Tests/Server_KeepAlive_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_KeepAlive_Tests.cs index 6d9a0208..decbb763 100644 --- a/Tests/MQTTnet.Core.Tests/Server_KeepAlive_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server_KeepAlive_Tests.cs @@ -21,6 +21,7 @@ namespace MQTTnet.Tests var client = await testEnvironment.ConnectLowLevelClientAsync(o => o .WithCommunicationTimeout(TimeSpan.FromSeconds(1)) + .WithCommunicationTimeout(TimeSpan.Zero) .WithProtocolVersion(MqttProtocolVersion.V500)).ConfigureAwait(false); await client.SendAsync(new MqttConnectPacket @@ -35,17 +36,18 @@ namespace MQTTnet.Tests for (var i = 0; i < 6; i++) { - await client.SendAsync(MqttPingReqPacket.Instance, CancellationToken.None); await Task.Delay(500); + + await client.SendAsync(MqttPingReqPacket.Instance, CancellationToken.None); responsePacket = await client.ReceiveAsync(CancellationToken.None); Assert.IsTrue(responsePacket is MqttPingRespPacket); } // If we reach this point everything works as expected (server did not close the connection // due to proper ping messages. - // Now we will wait 1.2 seconds because the server MUST wait 1.5 seconds in total (See spec). + // Now we will wait 1.1 seconds because the server MUST wait 1.5 seconds in total (See spec). - await Task.Delay(1200); + await Task.Delay(1100); await client.SendAsync(MqttPingReqPacket.Instance, CancellationToken.None); responsePacket = await client.ReceiveAsync(CancellationToken.None); Assert.IsTrue(responsePacket is MqttPingRespPacket); diff --git a/Tests/MQTTnet.Core.Tests/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Session_Tests.cs index 1f4c418d..8ca72ee6 100644 --- a/Tests/MQTTnet.Core.Tests/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Session_Tests.cs @@ -96,9 +96,7 @@ namespace MQTTnet.Tests var serverOptions = new MqttServerOptionsBuilder(); await testEnvironment.StartServerAsync(serverOptions); - var options = new MqttClientOptionsBuilder() - .WithClientId("1") - ; + var options = new MqttClientOptionsBuilder().WithClientId("1"); var clients = await Task.WhenAll(Enumerable.Range(0, 10) .Select(i => TryConnect(testEnvironment, options))); @@ -109,7 +107,7 @@ namespace MQTTnet.Tests } } - private async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) + async Task TryConnect(TestEnvironment testEnvironment, MqttClientOptionsBuilder options) { try {