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

Replace TaskCompletionSource in Client with Interlocked gate.

Browse Source
This commit is contained in:
Christian Kratky
2018-10-04 22:30:03 +02:00
parent 6bee6a0469
commit fb4f89b412
2 changed files with 17 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+11 -6
View File
@@ -29,7 +29,7 @@ namespace MQTTnet.Client
internal Task _keepAliveMessageSenderTask; internal Task _keepAliveMessageSenderTask;
private IMqttChannelAdapter _adapter; private IMqttChannelAdapter _adapter;
private bool _cleanDisconnectInitiated; private bool _cleanDisconnectInitiated;
private TaskCompletionSource<bool> _disconnectReason; private int _disconnectGate;
public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger) public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
{ {
@@ -59,7 +59,7 @@ namespace MQTTnet.Client
_packetDispatcher.Reset(); _packetDispatcher.Reset();
_cancellationTokenSource = new CancellationTokenSource(); _cancellationTokenSource = new CancellationTokenSource();
_disconnectReason = new TaskCompletionSource<bool>(); _disconnectGate = 0;
_adapter = _adapterFactory.CreateClientAdapter(options, _logger); _adapter = _adapterFactory.CreateClientAdapter(options, _logger);
_logger.Verbose($"Trying to connect with server ({_options.ChannelOptions})."); _logger.Verbose($"Trying to connect with server ({_options.ChannelOptions}).");
@@ -88,7 +88,7 @@ namespace MQTTnet.Client
{ {
_logger.Error(exception, "Error while connecting with server."); _logger.Error(exception, "Error while connecting with server.");
if (_disconnectReason.TrySetException(exception)) if (!DisconnectIsPending())
{ {
await DisconnectInternalAsync(null, exception).ConfigureAwait(false); await DisconnectInternalAsync(null, exception).ConfigureAwait(false);
} }
@@ -110,7 +110,7 @@ namespace MQTTnet.Client
} }
finally finally
{ {
if (_disconnectReason.TrySetCanceled()) if (!DisconnectIsPending())
{ {
await DisconnectInternalAsync(null, null).ConfigureAwait(false); await DisconnectInternalAsync(null, null).ConfigureAwait(false);
} }
@@ -361,7 +361,7 @@ namespace MQTTnet.Client
_logger.Error(exception, "Unhandled exception while sending/receiving keep alive packets."); _logger.Error(exception, "Unhandled exception while sending/receiving keep alive packets.");
} }
if (_disconnectReason.TrySetException(exception)) if (!DisconnectIsPending())
{ {
await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false); await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false);
} }
@@ -410,7 +410,7 @@ namespace MQTTnet.Client
_packetDispatcher.Dispatch(exception); _packetDispatcher.Dispatch(exception);
if (_disconnectReason.TrySetException(exception)) if (!DisconnectIsPending())
{ {
await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false); await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false);
} }
@@ -545,5 +545,10 @@ namespace MQTTnet.Client
{ {
} }
} }
private bool DisconnectIsPending()
{
return Interlocked.CompareExchange(ref _disconnectGate, 1, 0) != 0;
}
} }
} }
Tests/MQTTnet.Core.Tests/MqttServerTests.cs
+6 -7
View File
@@ -669,25 +669,24 @@ namespace MQTTnet.Core.Tests
MqttQualityOfServiceLevel filterQualityOfServiceLevel, MqttQualityOfServiceLevel filterQualityOfServiceLevel,
int expectedReceivedMessagesCount) int expectedReceivedMessagesCount)
{ {
var serverAdapter = new TestMqttServerAdapter(); var s = new MqttFactory().CreateMqttServer();
var s = new MqttFactory().CreateMqttServer(new[] { serverAdapter }, new MqttNetLogger());
var receivedMessagesCount = 0; var receivedMessagesCount = 0;
try try
{ {
await s.StartAsync(new MqttServerOptions()); await s.StartAsync(new MqttServerOptions());
var c1 = await serverAdapter.ConnectTestClient("c1"); var c1 = new MqttFactory().CreateMqttClient();
var c2 = await serverAdapter.ConnectTestClient("c2");
c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; c1.ApplicationMessageReceived += (_, __) => receivedMessagesCount++;
await c1.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build());
await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build()); await c1.SubscribeAsync(new TopicFilterBuilder().WithTopic(topicFilter).WithQualityOfServiceLevel(filterQualityOfServiceLevel).Build());
var c2 = new MqttFactory().CreateMqttClient();
await c2.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build());
await c2.PublishAsync(builder => builder.WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel)); await c2.PublishAsync(builder => builder.WithTopic(topic).WithPayload(new byte[0]).WithQualityOfServiceLevel(qualityOfServiceLevel));
await Task.Delay(500); await Task.Delay(500);
await c1.UnsubscribeAsync(topicFilter); await c1.UnsubscribeAsync(topicFilter);
await Task.Delay(500); await Task.Delay(500);
} }
finally finally