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

Refactor manage client

Browse Source
This commit is contained in:
Christian Kratky
2017-10-21 15:49:50 +02:00
parent d052f409b6
commit 24c35de330
17 changed files with 485 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.Core/ManagedClient/IManagedMqttClientOptions.cs
-1
View File
@@ -7,7 +7,6 @@ namespace MQTTnet.Core.ManagedClient
{
IMqttClientOptions ClientOptions { get; }
bool UseAutoReconnect { get; }
TimeSpan AutoReconnectDelay { get; }
IManagedMqttClientStorage Storage { get; }
MQTTnet.Core/ManagedClient/ManagedMqttClient.cs
+245 -88
View File
@@ -1,4 +1,5 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@@ -7,17 +8,24 @@ using MQTTnet.Core.Client;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
namespace MQTTnet.Core.ManagedClient
{
public class ManagedMqttClient
{
private readonly List<MqttApplicationMessage> _messageQueue = new List<MqttApplicationMessage>();
private readonly AutoResetEvent _messageQueueGate = new AutoResetEvent(false);
private readonly MqttClient _mqttClient;
private readonly ManagedMqttClientStorageManager _storageManager = new ManagedMqttClientStorageManager();
private readonly BlockingCollection<MqttApplicationMessage> _messageQueue = new BlockingCollection<MqttApplicationMessage>();
private readonly HashSet<TopicFilter> _subscriptions = new HashSet<TopicFilter>();
private readonly IMqttClient _mqttClient;
private readonly MqttNetTrace _trace;
private CancellationTokenSource _connectionCancellationToken;
private CancellationTokenSource _publishingCancellationToken;
private IManagedMqttClientOptions _options;
private bool _subscriptionsNotPushed;
public ManagedMqttClient(IMqttCommunicationAdapterFactory communicationChannelFactory, MqttNetTrace trace)
{
@@ -30,120 +38,269 @@ namespace MQTTnet.Core.ManagedClient
_mqttClient.ApplicationMessageReceived += OnApplicationMessageReceived;
}
public event EventHandler Connected;
public event EventHandler<MqttClientConnectedEventArgs> Connected;
public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public bool IsConnected => _mqttClient.IsConnected;
public void Start(IManagedMqttClientOptions options)
public async Task StartAsync(IManagedMqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (options.ClientOptions == null) throw new ArgumentException("The client options are not set.", nameof(options));
if (!options.ClientOptions.CleanSession)
{
throw new NotSupportedException("The managed client does not support existing sessions.");
}
}
if (_connectionCancellationToken != null)
{
throw new InvalidOperationException("The managed client is already started.");
}
public void Stop()
{
_options = options;
await _storageManager.SetStorageAsync(_options.Storage).ConfigureAwait(false);
if (_options.Storage != null)
{
var loadedMessages = await _options.Storage.LoadQueuedMessagesAsync().ConfigureAwait(false);
foreach (var loadedMessage in loadedMessages)
{
_messageQueue.Add(loadedMessage);
}
}
_connectionCancellationToken = new CancellationTokenSource();
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Factory.StartNew(() => MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
_trace.Information(nameof(ManagedMqttClient), "Started");
}
public async Task ConnectAsync(IManagedMqttClientOptions options)
public Task StopAsync()
{
//////TODO VERY BAD
////_options = options as ManagedMqttClientTcpOptions;
////this._usePersistance = _options.Storage != null;
////await _mqttClient.ConnectAsync(options);
////SetupOutgoingPacketProcessingAsync();
_connectionCancellationToken?.Cancel(false);
_connectionCancellationToken = null;
//////load persistentMessages
////////if (_usePersistance)
////////{
//////// if (_persistentMessagesManager == null)
//////// _persistentMessagesManager = new ManagedMqttClientMessagesManager(_options);
//////// await _persistentMessagesManager.LoadMessagesAsync();
//////// await InternalPublishAsync(_persistentMessagesManager.GetMessages(), false);
////////}
while (_messageQueue.Any())
{
_messageQueue.Take();
}
return Task.FromResult(0);
}
public async Task DisconnectAsync()
{
await _mqttClient.DisconnectAsync();
}
public async Task UnsubscribeAsync(IEnumerable<string> topicFilters)
{
// TODO: Move all subscriptions to list an subscribe after connection has lost. But only if server session is new.
await _mqttClient.UnsubscribeAsync(topicFilters);
}
public void Enqueue(IEnumerable<MqttApplicationMessage> applicationMessages)
public Task EnqueueAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
ThrowIfNotConnected();
foreach (var applicationMessage in applicationMessages)
{
_messageQueue.Add(applicationMessage);
}
_messageQueue.AddRange(applicationMessages);
_options.Storage?.SaveQueuedMessagesAsync(_messageQueue.ToList());
_messageQueueGate.Set();
}
public async Task<IList<MqttSubscribeResult>> SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
{
return await _mqttClient.SubscribeAsync(topicFilters);
return Task.FromResult(0);
}
private void ThrowIfNotConnected()
public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
{
if (!IsConnected) throw new MqttCommunicationException("The client is not connected.");
}
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
private void SetupOutgoingPacketProcessingAsync()
{
//Task.Factory.StartNew(
// () => SendPackets(_mqttClient._cancellationTokenSource.Token),
// _mqttClient._cancellationTokenSource.Token,
// TaskCreationOptions.LongRunning,
// TaskScheduler.Default).ConfigureAwait(false);
lock (_subscriptions)
{
foreach (var topicFilter in topicFilters)
{
if (_subscriptions.Add(topicFilter))
{
_subscriptionsNotPushed = true;
}
}
}
return Task.FromResult(0);
}
private async Task SendPackets(CancellationToken cancellationToken)
public Task UnsubscribeAsync(IEnumerable<TopicFilter> topicFilters)
{
//MqttNetTrace.Information(nameof(MqttClientManaged), "Start sending packets.");
//MqttApplicationMessage messageInQueue = null;
lock (_subscriptions)
{
foreach (var topicFilter in topicFilters)
{
if (_subscriptions.Remove(topicFilter))
{
_subscriptionsNotPushed = true;
}
}
}
//try
//{
// while (!cancellationToken.IsCancellationRequested)
// {
// messageInQueue = _inflightQueue.Take();
// await _mqttClient.PublishAsync(new List<MqttApplicationMessage>() { messageInQueue });
// if (_usePersistance)
// await _persistentMessagesManager.Remove(messageInQueue);
// }
//}
//catch (OperationCanceledException)
//{
//}
//catch (MqttCommunicationException exception)
//{
// MqttNetTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while sending packets.");
// //message not send, equeue it again
// if (messageInQueue != null)
// _inflightQueue.Add(messageInQueue);
//}
//catch (Exception exception)
//{
// MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while sending packets.");
// await DisconnectAsync().ConfigureAwait(false);
//}
//finally
//{
// MqttNetTrace.Information(nameof(MqttClient), "Stopped sending packets.");
//}
return Task.FromResult(0);
}
private async Task MaintainConnectionAsync(CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false);
if (connectionState == ReconnectionResult.NotConnected)
{
_publishingCancellationToken?.Cancel(false);
_publishingCancellationToken = null;
await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
continue;
}
if (connectionState == ReconnectionResult.Reconnected || _subscriptionsNotPushed)
{
await PushSubscriptionsAsync();
_publishingCancellationToken = new CancellationTokenSource();
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Factory.StartNew(() => PublishQueuedMessagesAsync(_publishingCancellationToken.Token), _publishingCancellationToken.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
continue;
}
if (connectionState == ReconnectionResult.StillConnected)
{
await Task.Delay(100, _connectionCancellationToken.Token).ConfigureAwait(false); // Consider using the _Disconnected_ event here. (TaskCompletionSource)
}
}
}
catch (OperationCanceledException)
{
}
catch (MqttCommunicationException exception)
{
_trace.Warning(nameof(ManagedMqttClient), exception, "Communication exception while maintaining connection.");
}
catch (Exception exception)
{
_trace.Error(nameof(ManagedMqttClient), exception, "Unhandled exception while maintaining connection.");
}
finally
{
await _mqttClient.DisconnectAsync().ConfigureAwait(false);
_trace.Information(nameof(ManagedMqttClient), "Stopped");
}
}
private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
{
try
{
while (!cancellationToken.IsCancellationRequested)
{
var message = _messageQueue.Take(cancellationToken);
if (message == null)
{
continue;
}
if (cancellationToken.IsCancellationRequested)
{
continue;
}
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
}
finally
{
_trace.Information(nameof(ManagedMqttClient), "Stopped publishing messages");
}
}
private async Task TryPublishQueuedMessageAsync(MqttApplicationMessage message)
{
try
{
await _mqttClient.PublishAsync(message).ConfigureAwait(false);
}
catch (MqttCommunicationException exception)
{
_trace.Warning(nameof(ManagedMqttClient), exception, "Publishing application message failed.");
if (message.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
_messageQueue.Add(message);
}
}
catch (Exception exception)
{
_trace.Error(nameof(ManagedMqttClient), exception, "Unhandled exception while publishing queued application message.");
}
}
private async Task SaveAsync(List<MqttApplicationMessage> messages)
{
if (messages == null)
{
return;
}
var storage = _options.Storage;
if (storage != null)
{
return;
}
await _options.Storage.SaveQueuedMessagesAsync(messages);
}
private async Task PushSubscriptionsAsync()
{
_trace.Information(nameof(ManagedMqttClient), "Synchronizing subscriptions");
List<TopicFilter> subscriptions;
lock (_subscriptions)
{
subscriptions = _subscriptions.ToList();
_subscriptionsNotPushed = false;
}
if (!_subscriptions.Any())
{
return;
}
try
{
await _mqttClient.SubscribeAsync(subscriptions).ConfigureAwait(false);
}
catch (Exception exception)
{
_trace.Warning(nameof(ManagedMqttClient), exception, "Synchronizing subscriptions failed");
_subscriptionsNotPushed = true;
}
}
private async Task<ReconnectionResult> ReconnectIfRequiredAsync()
{
if (_mqttClient.IsConnected)
{
return ReconnectionResult.StillConnected;
}
try
{
await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false);
return ReconnectionResult.Reconnected;
}
catch (Exception)
{
return ReconnectionResult.NotConnected;
}
}
private void OnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs eventArgs)
@@ -156,7 +313,7 @@ namespace MQTTnet.Core.ManagedClient
Disconnected?.Invoke(this, eventArgs);
}
private void OnConnected(object sender, EventArgs eventArgs)
private void OnConnected(object sender, MqttClientConnectedEventArgs eventArgs)
{
Connected?.Invoke(this, eventArgs);
}
MQTTnet.Core/ManagedClient/ManagedMqttClientExtensions.cs
+30
View File
@@ -0,0 +1,30 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Core.Packets;
namespace MQTTnet.Core.ManagedClient
{
public static class ManagedMqttClientExtensions
{
public static Task EnqueueAsync(this ManagedMqttClient managedClient, params MqttApplicationMessage[] applicationMessages)
{
if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
return managedClient.EnqueueAsync(applicationMessages);
}
public static Task SubscribeAsync(this ManagedMqttClient managedClient, params TopicFilter[] topicFilters)
{
if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
return managedClient.SubscribeAsync(topicFilters);
}
public static Task UnsubscribeAsync(this ManagedMqttClient managedClient, params TopicFilter[] topicFilters)
{
if (managedClient == null) throw new ArgumentNullException(nameof(managedClient));
return managedClient.UnsubscribeAsync(topicFilters);
}
}
}
MQTTnet.Core/ManagedClient/ManagedMqttClientOptions.cs
-1
View File
@@ -7,7 +7,6 @@ namespace MQTTnet.Core.ManagedClient
{
public IMqttClientOptions ClientOptions { get; set; }
public bool UseAutoReconnect { get; set; } = true;
public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
public IManagedMqttClientStorage Storage { get; set; }
MQTTnet.Core/ManagedClient/ManagedMqttClientStorageManager.cs
+75
View File
@@ -0,0 +1,75 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
namespace MQTTnet.Core.ManagedClient
{
public class ManagedMqttClientStorageManager
{
private readonly List<MqttApplicationMessage> _applicationMessages = new List<MqttApplicationMessage>();
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private IManagedMqttClientStorage _storage;
public async Task SetStorageAsync(IManagedMqttClientStorage storage)
{
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
_storage = storage;
}
finally
{
_semaphore.Release();
}
}
public async Task AddAsync(MqttApplicationMessage applicationMessage)
{
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
if (_storage == null)
{
return;
}
_applicationMessages.Add(applicationMessage);
await SaveAsync().ConfigureAwait(false);
}
finally
{
_semaphore.Release();
}
}
public async Task RemoveAsync(MqttApplicationMessage applicationMessage)
{
await _semaphore.WaitAsync().ConfigureAwait(false);
try
{
if (_storage == null)
{
return;
}
var index = _applicationMessages.IndexOf(applicationMessage);
if (index == -1)
{
return;
}
_applicationMessages.RemoveAt(index);
await SaveAsync().ConfigureAwait(false);
}
finally
{
_semaphore.Release();
}
}
private Task SaveAsync()
{
return _storage.SaveQueuedMessagesAsync(_applicationMessages);
}
}
}
MQTTnet.Core/ManagedClient/ReconnectionResult.cs
+9
View File
@@ -0,0 +1,9 @@
namespace MQTTnet.Core.ManagedClient
{
public enum ReconnectionResult
{
StillConnected,
Reconnected,
NotConnected
}
}