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

Refactor code.

Browse Source
This commit is contained in:
Christian
2020-12-08 21:07:58 +01:00
parent 1214b15260
commit b0699003e2
3 changed files with 23 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+21 -22
View File
@@ -32,7 +32,6 @@ namespace MQTTnet.Extensions.ManagedClient
readonly HashSet<string> _unsubscriptions = new HashSet<string>(); readonly HashSet<string> _unsubscriptions = new HashSet<string>();
readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0); readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
readonly IMqttClient _mqttClient;
readonly IMqttNetScopedLogger _logger; readonly IMqttNetScopedLogger _logger;
readonly AsyncLock _messageQueueLock = new AsyncLock(); readonly AsyncLock _messageQueueLock = new AsyncLock();
@@ -45,14 +44,13 @@ namespace MQTTnet.Extensions.ManagedClient
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger) public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
{ {
_mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient)); InternalClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
InternalClient = mqttClient;
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateScopedLogger(nameof(ManagedMqttClient)); _logger = logger.CreateScopedLogger(nameof(ManagedMqttClient));
} }
public bool IsConnected => _mqttClient.IsConnected; public bool IsConnected => InternalClient.IsConnected;
public bool IsStarted => _connectionCancellationToken != null; public bool IsStarted => _connectionCancellationToken != null;
@@ -64,20 +62,20 @@ namespace MQTTnet.Extensions.ManagedClient
public IMqttClientConnectedHandler ConnectedHandler public IMqttClientConnectedHandler ConnectedHandler
{ {
get => _mqttClient.ConnectedHandler; get => InternalClient.ConnectedHandler;
set => _mqttClient.ConnectedHandler = value; set => InternalClient.ConnectedHandler = value;
} }
public IMqttClientDisconnectedHandler DisconnectedHandler public IMqttClientDisconnectedHandler DisconnectedHandler
{ {
get => _mqttClient.DisconnectedHandler; get => InternalClient.DisconnectedHandler;
set => _mqttClient.DisconnectedHandler = value; set => InternalClient.DisconnectedHandler = value;
} }
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
{ {
get => _mqttClient.ApplicationMessageReceivedHandler; get => InternalClient.ApplicationMessageReceivedHandler;
set => _mqttClient.ApplicationMessageReceivedHandler = value; set => InternalClient.ApplicationMessageReceivedHandler = value;
} }
public IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; } public IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; }
@@ -138,7 +136,7 @@ namespace MQTTnet.Extensions.ManagedClient
public Task PingAsync(CancellationToken cancellationToken) public Task PingAsync(CancellationToken cancellationToken)
{ {
return _mqttClient.PingAsync(cancellationToken); return InternalClient.PingAsync(cancellationToken);
} }
public async Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) public async Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
@@ -265,7 +263,7 @@ namespace MQTTnet.Extensions.ManagedClient
_messageQueue.Dispose(); _messageQueue.Dispose();
_messageQueueLock.Dispose(); _messageQueueLock.Dispose();
_mqttClient.Dispose(); InternalClient.Dispose();
_subscriptionsQueuedSignal.Dispose(); _subscriptionsQueuedSignal.Dispose();
} }
@@ -294,7 +292,7 @@ namespace MQTTnet.Extensions.ManagedClient
{ {
try try
{ {
await _mqttClient.DisconnectAsync().ConfigureAwait(false); await InternalClient.DisconnectAsync().ConfigureAwait(false);
} }
catch (Exception exception) catch (Exception exception)
{ {
@@ -318,7 +316,7 @@ namespace MQTTnet.Extensions.ManagedClient
{ {
try try
{ {
var connectionState = await ReconnectIfRequiredAsync().ConfigureAwait(false); var connectionState = await ReconnectIfRequiredAsync(cancellationToken).ConfigureAwait(false);
if (connectionState == ReconnectionResult.NotConnected) if (connectionState == ReconnectionResult.NotConnected)
{ {
StopPublishing(); StopPublishing();
@@ -361,7 +359,7 @@ namespace MQTTnet.Extensions.ManagedClient
{ {
try try
{ {
while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected) while (!cancellationToken.IsCancellationRequested && InternalClient.IsConnected)
{ {
// Peek at the message without dequeueing in order to prevent the // Peek at the message without dequeueing in order to prevent the
// possibility of the queue growing beyond the configured cap. // possibility of the queue growing beyond the configured cap.
@@ -400,7 +398,7 @@ namespace MQTTnet.Extensions.ManagedClient
Exception transmitException = null; Exception transmitException = null;
try try
{ {
await _mqttClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false); await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync using (await _messageQueueLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync
{ {
@@ -463,6 +461,7 @@ namespace MQTTnet.Extensions.ManagedClient
async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken) async Task PublishSubscriptionsAsync(TimeSpan timeout, CancellationToken cancellationToken)
{ {
var endTime = DateTime.UtcNow + timeout; var endTime = DateTime.UtcNow + timeout;
while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false)) while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false))
{ {
List<MqttTopicFilter> subscriptions; List<MqttTopicFilter> subscriptions;
@@ -497,12 +496,12 @@ namespace MQTTnet.Extensions.ManagedClient
{ {
if (unsubscriptions.Any()) if (unsubscriptions.Any())
{ {
await _mqttClient.UnsubscribeAsync(unsubscriptions.ToArray()).ConfigureAwait(false); await InternalClient.UnsubscribeAsync(unsubscriptions.ToArray()).ConfigureAwait(false);
} }
if (subscriptions.Any()) if (subscriptions.Any())
{ {
await _mqttClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false); await InternalClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false);
} }
} }
catch (Exception exception) catch (Exception exception)
@@ -521,7 +520,7 @@ namespace MQTTnet.Extensions.ManagedClient
if (_reconnectSubscriptions.Any()) if (_reconnectSubscriptions.Any())
{ {
var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }); var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value });
await _mqttClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false); await InternalClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false);
} }
} }
catch (Exception exception) catch (Exception exception)
@@ -541,16 +540,16 @@ namespace MQTTnet.Extensions.ManagedClient
} }
} }
async Task<ReconnectionResult> ReconnectIfRequiredAsync() async Task<ReconnectionResult> ReconnectIfRequiredAsync(CancellationToken cancellationToken)
{ {
if (_mqttClient.IsConnected) if (InternalClient.IsConnected)
{ {
return ReconnectionResult.StillConnected; return ReconnectionResult.StillConnected;
} }
try try
{ {
var result = await _mqttClient.ConnectAsync(Options.ClientOptions).ConfigureAwait(false); var result = await InternalClient.ConnectAsync(Options.ClientOptions, cancellationToken).ConfigureAwait(false);
return result.IsSessionPresent ? ReconnectionResult.Recovered : ReconnectionResult.Reconnected; return result.IsSessionPresent ? ReconnectionResult.Recovered : ReconnectionResult.Reconnected;
} }
catch (Exception exception) catch (Exception exception)
Source/MQTTnet/Client/MqttClient.cs
+1 -1
View File
@@ -90,7 +90,7 @@ namespace MQTTnet.Client
using (var combined = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken)) using (var combined = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken))
{ {
_logger.Verbose("Trying to connect with server '{0}' (Timeout={1}).", options.ChannelOptions, options.CommunicationTimeout); _logger.Verbose("Trying to connect with server '{0}' (Timeout={1}).", options.ChannelOptions, options.CommunicationTimeout);
await _adapter.ConnectAsync(options.CommunicationTimeout, combined.Token).ConfigureAwait(false); await adapter.ConnectAsync(options.CommunicationTimeout, combined.Token).ConfigureAwait(false);
_logger.Verbose("Connection with server established."); _logger.Verbose("Connection with server established.");
_publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>(); _publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
Source/MQTTnet/Implementations/CrossPlatformSocket.cs
+1 -2
View File
@@ -72,11 +72,10 @@ namespace MQTTnet.Implementations
{ {
#if NET452 || NET461 #if NET452 || NET461
var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false); var clientSocket = await Task.Factory.FromAsync(_socket.BeginAccept, _socket.EndAccept, null).ConfigureAwait(false);
return new CrossPlatformSocket(clientSocket);
#else #else
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false); var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
return new CrossPlatformSocket(clientSocket);
#endif #endif
return new CrossPlatformSocket(clientSocket);
} }
catch (ObjectDisposedException) catch (ObjectDisposedException)
{ {