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

Refactor task usage in client.

Browse Source
This commit is contained in:
Christian Kratky
2018-11-17 20:23:52 +01:00
parent af39e87703
commit 5f18f35e02
1 changed files with 11 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+11 -11
View File
@@ -139,7 +139,7 @@ namespace MQTTnet.Client
return subscribePacket.TopicFilters.Select((t, i) => new MqttSubscribeResult(t, response.SubscribeReturnCodes[i])).ToList(); return subscribePacket.TopicFilters.Select((t, i) => new MqttSubscribeResult(t, response.SubscribeReturnCodes[i])).ToList();
} }
public async Task UnsubscribeAsync(IEnumerable<string> topicFilters) public Task UnsubscribeAsync(IEnumerable<string> topicFilters)
{ {
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
@@ -151,7 +151,7 @@ namespace MQTTnet.Client
TopicFilters = topicFilters.ToList() TopicFilters = topicFilters.ToList()
}; };
await SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false); return SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket, _cancellationTokenSource.Token);
} }
public Task PublishAsync(MqttApplicationMessage applicationMessage) public Task PublishAsync(MqttApplicationMessage applicationMessage)
@@ -234,14 +234,14 @@ namespace MQTTnet.Client
try try
{ {
await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false);
await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false);
if (_adapter != null) if (_adapter != null)
{ {
await _adapter.DisconnectAsync(_options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false); await _adapter.DisconnectAsync(_options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
} }
await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false);
await WaitForTaskAsync(_keepAliveMessageSenderTask, sender).ConfigureAwait(false);
_logger.Verbose("Disconnected from adapter."); _logger.Verbose("Disconnected from adapter.");
} }
catch (Exception adapterException) catch (Exception adapterException)
@@ -320,10 +320,10 @@ namespace MQTTnet.Client
private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken) private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
{ {
_logger.Verbose("Start sending keep alive packets.");
try try
{ {
_logger.Verbose("Start sending keep alive packets.");
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
var keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75); var keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75);
@@ -374,10 +374,10 @@ namespace MQTTnet.Client
private async Task ReceivePacketsAsync(CancellationToken cancellationToken) private async Task ReceivePacketsAsync(CancellationToken cancellationToken)
{ {
_logger.Verbose("Start receiving packets.");
try try
{ {
_logger.Verbose("Start receiving packets.");
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken) var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken)
@@ -500,7 +500,7 @@ namespace MQTTnet.Client
() => ReceivePacketsAsync(cancellationToken), () => ReceivePacketsAsync(cancellationToken),
cancellationToken, cancellationToken,
TaskCreationOptions.LongRunning, TaskCreationOptions.LongRunning,
TaskScheduler.Default); TaskScheduler.Default).Unwrap();
} }
private void StartSendingKeepAliveMessages(CancellationToken cancellationToken) private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
@@ -509,7 +509,7 @@ namespace MQTTnet.Client
() => SendKeepAliveMessagesAsync(cancellationToken), () => SendKeepAliveMessagesAsync(cancellationToken),
cancellationToken, cancellationToken,
TaskCreationOptions.LongRunning, TaskCreationOptions.LongRunning,
TaskScheduler.Default); TaskScheduler.Default).Unwrap();
} }
private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)