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

corrected thread safety of ping timers after adding delayed acknowledgement

Browse Source
This commit is contained in:
tamas.kurucsai
2021-01-29 07:03:56 +01:00
parent 822f582e41
commit fbf7d9bb5c
1 changed files with 8 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+8 -8
View File
@@ -25,8 +25,8 @@ namespace MQTTnet.Client
{ {
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
readonly Stopwatch _sendTracker = new Stopwatch(); volatile int _lastSentAtTicks;
readonly Stopwatch _receiveTracker = new Stopwatch(); volatile int _lastReceivedAtTicks;
readonly object _disconnectLock = new object(); readonly object _disconnectLock = new object();
readonly IMqttClientAdapterFactory _adapterFactory; readonly IMqttClientAdapterFactory _adapterFactory;
@@ -102,8 +102,8 @@ namespace MQTTnet.Client
authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false); authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false);
} }
_sendTracker.Restart(); _lastSentAtTicks = Environment.TickCount;
_receiveTracker.Restart(); _lastReceivedAtTicks = Environment.TickCount;
if (Options.KeepAlivePeriod != TimeSpan.Zero) if (Options.KeepAlivePeriod != TimeSpan.Zero)
{ {
@@ -382,7 +382,7 @@ namespace MQTTnet.Client
{ {
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
_sendTracker.Restart(); _lastReceivedAtTicks = Environment.TickCount;
return _adapter.SendPacketAsync(packet, cancellationToken); return _adapter.SendPacketAsync(packet, cancellationToken);
} }
@@ -401,7 +401,7 @@ namespace MQTTnet.Client
{ {
try try
{ {
_sendTracker.Restart(); _lastSentAtTicks = Environment.TickCount;
await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false); await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
} }
catch (Exception e) catch (Exception e)
@@ -437,7 +437,7 @@ namespace MQTTnet.Client
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
// Values described here: [MQTT-3.1.2-24]. // Values described here: [MQTT-3.1.2-24].
var waitTime = keepAlivePeriod - _sendTracker.Elapsed; var waitTime = keepAlivePeriod - TimeSpan.FromMilliseconds(unchecked(Environment.TickCount - _lastSentAtTicks));
if (waitTime <= TimeSpan.Zero) if (waitTime <= TimeSpan.Zero)
{ {
@@ -546,7 +546,7 @@ namespace MQTTnet.Client
{ {
try try
{ {
_receiveTracker.Restart(); _lastReceivedAtTicks = Environment.TickCount;
if (packet is MqttPublishPacket publishPacket) if (packet is MqttPublishPacket publishPacket)
{ {