From 2a421d1d428933ac435206f0b1a46f420f1687dd Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Fri, 1 May 2020 21:47:25 +0200 Subject: [PATCH] Refactor code. --- Source/MQTTnet/Client/MqttClient.cs | 2 +- .../Internal/ArraySegmentExtensions.cs | 23 -------------- Source/MQTTnet/MQTTnet.csproj | 6 +--- .../PacketDispatcher/IMqttPacketAwaiter.cs | 4 +-- .../PacketDispatcher/MqttPacketAwaiter.cs | 19 ++++-------- .../PacketDispatcher/MqttPacketDispatcher.cs | 30 +++++++++---------- Source/MQTTnet/Server/MqttClientConnection.cs | 6 ++-- Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs | 10 +++++-- 8 files changed, 36 insertions(+), 64 deletions(-) delete mode 100644 Source/MQTTnet/Internal/ArraySegmentExtensions.cs diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index f81cd3d8..4c71e6bb 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -400,7 +400,7 @@ namespace MQTTnet.Client identifier = packetWithIdentifier.PacketIdentifier.Value; } - using (var packetAwaiter = _packetDispatcher.AddPacketAwaiter(identifier)) + using (var packetAwaiter = _packetDispatcher.AddAwaiter(identifier)) { try { diff --git a/Source/MQTTnet/Internal/ArraySegmentExtensions.cs b/Source/MQTTnet/Internal/ArraySegmentExtensions.cs deleted file mode 100644 index f0054f47..00000000 --- a/Source/MQTTnet/Internal/ArraySegmentExtensions.cs +++ /dev/null @@ -1,23 +0,0 @@ -using System; - -namespace MQTTnet.Internal -{ - public static class ArraySegmentExtensions - { - public static byte[] ToArray(this ArraySegment source) - { - if (source.Array == null) - { - return null; - } - - var buffer = new byte[source.Count]; - if (buffer.Length > 0) - { - Array.Copy(source.Array, source.Offset, buffer, 0, buffer.Length); - } - - return buffer; - } - } -} diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 896f7a86..f6f52d67 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -54,7 +54,7 @@ - + @@ -62,9 +62,5 @@ - - - - \ No newline at end of file diff --git a/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs b/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs index 786dc97d..c1c2d4eb 100644 --- a/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs +++ b/Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs @@ -1,5 +1,5 @@ -using System; -using MQTTnet.Packets; +using MQTTnet.Packets; +using System; namespace MQTTnet.PacketDispatcher { diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs index d8f0a043..0ae23155 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs @@ -1,5 +1,4 @@ using MQTTnet.Exceptions; -using MQTTnet.Internal; using MQTTnet.Packets; using System; using System.Threading; @@ -7,11 +6,11 @@ using System.Threading.Tasks; namespace MQTTnet.PacketDispatcher { - public sealed class MqttPacketAwaiter : Disposable, IMqttPacketAwaiter where TPacket : MqttBasePacket + public sealed class MqttPacketAwaiter : IMqttPacketAwaiter where TPacket : MqttBasePacket { - private readonly TaskCompletionSource _taskCompletionSource; - private readonly ushort? _packetIdentifier; - private readonly MqttPacketDispatcher _owningPacketDispatcher; + readonly TaskCompletionSource _taskCompletionSource; + readonly ushort? _packetIdentifier; + readonly MqttPacketDispatcher _owningPacketDispatcher; public MqttPacketAwaiter(ushort? packetIdentifier, MqttPacketDispatcher owningPacketDispatcher) { @@ -39,7 +38,6 @@ namespace MQTTnet.PacketDispatcher { if (packet == null) throw new ArgumentNullException(nameof(packet)); - #if NET452 // To prevent deadlocks it is required to call the _TrySetResult_ method // from a new thread because the awaiting code will not(!) be executed in @@ -81,14 +79,9 @@ namespace MQTTnet.PacketDispatcher #endif } - protected override void Dispose(bool disposing) + public void Dispose() { - if (disposing) - { - _owningPacketDispatcher.RemovePacketAwaiter(_packetIdentifier); - } - - base.Dispose(disposing); + _owningPacketDispatcher.RemoveAwaiter(_packetIdentifier); } } } \ No newline at end of file diff --git a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs index ef245e90..785fd353 100644 --- a/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs +++ b/Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs @@ -1,22 +1,22 @@ -using System; -using System.Collections.Concurrent; using MQTTnet.Exceptions; using MQTTnet.Packets; +using System; +using System.Collections.Concurrent; namespace MQTTnet.PacketDispatcher { - public class MqttPacketDispatcher + public sealed class MqttPacketDispatcher { - private readonly ConcurrentDictionary, IMqttPacketAwaiter> _packetAwaiters = new ConcurrentDictionary, IMqttPacketAwaiter>(); + readonly ConcurrentDictionary, IMqttPacketAwaiter> _awaiters = new ConcurrentDictionary, IMqttPacketAwaiter>(); public void Dispatch(Exception exception) { - foreach (var awaiter in _packetAwaiters) + foreach (var awaiter in _awaiters) { awaiter.Value.Fail(exception); } - _packetAwaiters.Clear(); + _awaiters.Clear(); } public void Dispatch(MqttBasePacket packet) @@ -25,7 +25,7 @@ namespace MQTTnet.PacketDispatcher if (packet is MqttDisconnectPacket disconnectPacket) { - foreach (var packetAwaiter in _packetAwaiters) + foreach (var packetAwaiter in _awaiters) { packetAwaiter.Value.Fail(new MqttUnexpectedDisconnectReceivedException(disconnectPacket)); } @@ -42,7 +42,7 @@ namespace MQTTnet.PacketDispatcher var type = packet.GetType(); var key = new Tuple(identifier, type); - if (_packetAwaiters.TryRemove(key, out var awaiter)) + if (_awaiters.TryRemove(key, out var awaiter)) { awaiter.Complete(packet); return; @@ -53,15 +53,15 @@ namespace MQTTnet.PacketDispatcher public void Reset() { - foreach (var awaiter in _packetAwaiters) + foreach (var awaiter in _awaiters) { - awaiter.Value.Cancel(); + awaiter.Value.Cancel(); } - _packetAwaiters.Clear(); + _awaiters.Clear(); } - public MqttPacketAwaiter AddPacketAwaiter(ushort? identifier) where TResponsePacket : MqttBasePacket + public MqttPacketAwaiter AddAwaiter(ushort? identifier) where TResponsePacket : MqttBasePacket { if (!identifier.HasValue) { @@ -71,7 +71,7 @@ namespace MQTTnet.PacketDispatcher var awaiter = new MqttPacketAwaiter(identifier, this); var key = new Tuple(identifier.Value, typeof(TResponsePacket)); - if (!_packetAwaiters.TryAdd(key, awaiter)) + if (!_awaiters.TryAdd(key, awaiter)) { throw new InvalidOperationException($"The packet dispatcher already has an awaiter for packet of type '{key.Item2.Name}' with identifier {key.Item1}."); } @@ -79,7 +79,7 @@ namespace MQTTnet.PacketDispatcher return awaiter; } - public void RemovePacketAwaiter(ushort? identifier) where TResponsePacket : MqttBasePacket + public void RemoveAwaiter(ushort? identifier) where TResponsePacket : MqttBasePacket { if (!identifier.HasValue) { @@ -87,7 +87,7 @@ namespace MQTTnet.PacketDispatcher } var key = new Tuple(identifier.Value, typeof(TResponsePacket)); - _packetAwaiters.TryRemove(key, out _); + _awaiters.TryRemove(key, out _); } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttClientConnection.cs b/Source/MQTTnet/Server/MqttClientConnection.cs index 43a86777..082082da 100644 --- a/Source/MQTTnet/Server/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/MqttClientConnection.cs @@ -430,14 +430,14 @@ namespace MQTTnet.Server } else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) { - var awaiter = _packetDispatcher.AddPacketAwaiter(publishPacket.PacketIdentifier); + var awaiter = _packetDispatcher.AddAwaiter(publishPacket.PacketIdentifier); await SendAsync(publishPacket).ConfigureAwait(false); await awaiter.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); } else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) { - using (var awaiter1 = _packetDispatcher.AddPacketAwaiter(publishPacket.PacketIdentifier)) - using (var awaiter2 = _packetDispatcher.AddPacketAwaiter(publishPacket.PacketIdentifier)) + using (var awaiter1 = _packetDispatcher.AddAwaiter(publishPacket.PacketIdentifier)) + using (var awaiter2 = _packetDispatcher.AddAwaiter(publishPacket.PacketIdentifier)) { await SendAsync(publishPacket).ConfigureAwait(false); await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false); diff --git a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs index 2924e8e5..774de395 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs @@ -36,7 +36,7 @@ namespace MQTTnet.Tests public async Task Connect_To_Invalid_Server_Port_Not_Opened() { var client = new MqttFactory().CreateMqttClient(); - await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", 12345).WithCommunicationTimeout(TimeSpan.FromSeconds(2)).Build()); + await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", 12345).WithCommunicationTimeout(TimeSpan.FromSeconds(5)).Build()); } [TestMethod] @@ -115,7 +115,13 @@ namespace MQTTnet.Tests } }); - client2.UseApplicationMessageReceivedHandler(async c => { await client2.PublishAsync("reply", null, MqttQualityOfServiceLevel.AtLeastOnce); }); + client2.UseApplicationMessageReceivedHandler(async c => + { + // Use AtMostOnce here because with QoS 1 or even QoS 2 the process waits for + // the ACK etc. The problem is that the SpinUntil below only waits until the + // flag is set. It does not wait until the client has sent the ACK + await client2.PublishAsync("reply", null, MqttQualityOfServiceLevel.AtMostOnce); + }); await client1.PublishAsync("request", null, MqttQualityOfServiceLevel.AtLeastOnce);