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

Merge pull request #852 from cdobler/check-"Keep-Alive"-while-other-messages-are-sent-#851

Browse Source 
Added receiveTracker to detect whether messages are received from broker.
This commit is contained in:
HansM
2020-02-24 17:02:42 +01:00
committed by GitHub
parent 4d27f42e72 122034296c
commit 76ddf8c503
1 changed files with 13 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+13 -7
View File
@@ -1,4 +1,4 @@
using System;
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
@@ -25,6 +25,7 @@ namespace MQTTnet.Client
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
private readonly Stopwatch _sendTracker = new Stopwatch();
private readonly Stopwatch _receiveTracker = new Stopwatch();
private readonly object _disconnectLock = new object();
private readonly IMqttClientAdapterFactory _adapterFactory;
@@ -93,6 +94,7 @@ namespace MQTTnet.Client
}
_sendTracker.Restart();
_receiveTracker.Restart();
if (Options.KeepAlivePeriod != TimeSpan.Zero)
{
@@ -381,7 +383,9 @@ namespace MQTTnet.Client
try
{
return await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false);
var response = await packetAwaiter.WaitOneAsync(Options.CommunicationTimeout).ConfigureAwait(false);
_receiveTracker.Restart();
return response;
}
catch (Exception exception)
{
@@ -410,14 +414,14 @@ namespace MQTTnet.Client
keepAliveSendInterval = Options.KeepAliveSendInterval.Value;
}
var waitTime = keepAliveSendInterval - _sendTracker.Elapsed;
if (waitTime <= TimeSpan.Zero)
var waitTimeSend = keepAliveSendInterval - _sendTracker.Elapsed;
var waitTimeReceive = keepAliveSendInterval - _receiveTracker.Elapsed;
if (waitTimeSend <= TimeSpan.Zero || waitTimeReceive <= TimeSpan.Zero)
{
await SendAndReceiveAsync<MqttPingRespPacket>(new MqttPingReqPacket(), cancellationToken).ConfigureAwait(false);
waitTime = keepAliveSendInterval;
}
await Task.Delay(waitTime, cancellationToken).ConfigureAwait(false);
await Task.Delay(keepAliveSendInterval, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception exception)
@@ -514,6 +518,8 @@ namespace MQTTnet.Client
{
try
{
_receiveTracker.Restart();
if (packet is MqttPublishPacket publishPacket)
{
await TryProcessReceivedPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false);
@@ -703,4 +709,4 @@ namespace MQTTnet.Client
return Interlocked.CompareExchange(ref _disconnectGate, 1, 0) != 0;
}
}
}
}