YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-15 00:15:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor code.

Browse Source
This commit is contained in:
Christian Kratky
2020-05-01 21:47:25 +02:00
parent 48d0344a40
commit 2a421d1d42
8 changed files with 36 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+1 -1
View File
@@ -400,7 +400,7 @@ namespace MQTTnet.Client
identifier = packetWithIdentifier.PacketIdentifier.Value;
}
using (var packetAwaiter = _packetDispatcher.AddPacketAwaiter<TResponsePacket>(identifier))
using (var packetAwaiter = _packetDispatcher.AddAwaiter<TResponsePacket>(identifier))
{
try
{
Source/MQTTnet/Internal/ArraySegmentExtensions.cs
-23
View File
@@ -1,23 +0,0 @@
using System;
namespace MQTTnet.Internal
{
public static class ArraySegmentExtensions
{
public static byte[] ToArray(this ArraySegment<byte> 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;
}
}
}
Source/MQTTnet/MQTTnet.csproj
+1 -5
View File
@@ -54,7 +54,7 @@
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.8" />
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net452'">
@@ -62,9 +62,5 @@
<ItemGroup Condition="'$(TargetFramework)'=='net461'">
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.2.10" />
</ItemGroup>
</Project>
Source/MQTTnet/PacketDispatcher/IMqttPacketAwaiter.cs
+2 -2
View File
@@ -1,5 +1,5 @@
using System;
using MQTTnet.Packets;
using MQTTnet.Packets;
using System;
namespace MQTTnet.PacketDispatcher
{
Source/MQTTnet/PacketDispatcher/MqttPacketAwaiter.cs
+6 -13
View File
@@ -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<TPacket> : Disposable, IMqttPacketAwaiter where TPacket : MqttBasePacket
public sealed class MqttPacketAwaiter<TPacket> : IMqttPacketAwaiter where TPacket : MqttBasePacket
{
private readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource;
private readonly ushort? _packetIdentifier;
private readonly MqttPacketDispatcher _owningPacketDispatcher;
readonly TaskCompletionSource<MqttBasePacket> _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<TPacket>(_packetIdentifier);
}
base.Dispose(disposing);
_owningPacketDispatcher.RemoveAwaiter<TPacket>(_packetIdentifier);
}
}
}
Source/MQTTnet/PacketDispatcher/MqttPacketDispatcher.cs
+15 -15
View File
@@ -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<Tuple<ushort, Type>, IMqttPacketAwaiter> _packetAwaiters = new ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter>();
readonly ConcurrentDictionary<Tuple<ushort, Type>, IMqttPacketAwaiter> _awaiters = new ConcurrentDictionary<Tuple<ushort, Type>, 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<ushort, Type>(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<TResponsePacket> AddPacketAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
public MqttPacketAwaiter<TResponsePacket> AddAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
{
if (!identifier.HasValue)
{
@@ -71,7 +71,7 @@ namespace MQTTnet.PacketDispatcher
var awaiter = new MqttPacketAwaiter<TResponsePacket>(identifier, this);
var key = new Tuple<ushort, Type>(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<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
public void RemoveAwaiter<TResponsePacket>(ushort? identifier) where TResponsePacket : MqttBasePacket
{
if (!identifier.HasValue)
{
@@ -87,7 +87,7 @@ namespace MQTTnet.PacketDispatcher
}
var key = new Tuple<ushort, Type>(identifier.Value, typeof(TResponsePacket));
_packetAwaiters.TryRemove(key, out _);
_awaiters.TryRemove(key, out _);
}
}
}
Source/MQTTnet/Server/MqttClientConnection.cs
+3 -3
View File
@@ -430,14 +430,14 @@ namespace MQTTnet.Server
}
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
var awaiter = _packetDispatcher.AddPacketAwaiter<MqttPubAckPacket>(publishPacket.PacketIdentifier);
var awaiter = _packetDispatcher.AddAwaiter<MqttPubAckPacket>(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<MqttPubRecPacket>(publishPacket.PacketIdentifier))
using (var awaiter2 = _packetDispatcher.AddPacketAwaiter<MqttPubCompPacket>(publishPacket.PacketIdentifier))
using (var awaiter1 = _packetDispatcher.AddAwaiter<MqttPubRecPacket>(publishPacket.PacketIdentifier))
using (var awaiter2 = _packetDispatcher.AddAwaiter<MqttPubCompPacket>(publishPacket.PacketIdentifier))
{
await SendAsync(publishPacket).ConfigureAwait(false);
await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs
+8 -2
View File
@@ -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);