diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index 669f205c..6054370b 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -15,7 +15,7 @@ Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.Extensions.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index 6d7f7f69..ba5333f2 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -15,7 +15,7 @@ Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index d7eddb00..0df8e7e6 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -15,7 +15,7 @@ Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 556957d9..22cd3ae1 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -21,6 +21,9 @@ * [ManagedClient] The managed client is moved to a separate nuget package. * [ManagedClient] Added an own message format with extended properties like ID (BREAKING CHANGE). * [ManagedClient] Fixed a loading issue of stored application messages (thanks to @JTrotta). +* [ManagedClient] Added a new event which is fired when a synchronization of the subscriptions has failed. +* [ManagedClient] Added a new event which is fired when a connection attempt has failed. +* [ManagedClient] Exposed a new property which provides the count of not published messages (pending messages count). * [Server] Added support for other WebSocket sub protocol formats like mqttv-3.1.1 (thanks to @israellot). * [Server] The takeover of an existing client sessions is now treated as a _clean_ disconnect of the previous client. * [Server] The pending messages queue per client is now limited to 250 messages. Overflow strategy and count can be changed via options (thanks to @VladimirAkopyan) diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index e4eaa687..b5a6c50b 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -9,12 +9,16 @@ namespace MQTTnet.Extensions.ManagedClient { bool IsStarted { get; } bool IsConnected { get; } + int PendingApplicationMessagesCount { get; } event EventHandler Connected; event EventHandler Disconnected; event EventHandler ApplicationMessageProcessed; + event EventHandler ConnectingFailed; + event EventHandler SynchronizingSubscriptionsFailed; + Task StartAsync(IManagedMqttClientOptions options); Task StopAsync(); diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 687b0cdc..bcd98ae7 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -14,8 +14,8 @@ namespace MQTTnet.Extensions.ManagedClient public class ManagedMqttClient : IManagedMqttClient { private readonly BlockingCollection _messageQueue = new BlockingCollection(); - private readonly ConcurrentDictionary _subscriptions = new ConcurrentDictionary(); - private readonly List _unsubscriptions = new List(); + private readonly Dictionary _subscriptions = new Dictionary(); + private readonly HashSet _unsubscriptions = new HashSet(); private readonly IMqttClient _mqttClient; private readonly IMqttNetChildLogger _logger; @@ -43,12 +43,16 @@ namespace MQTTnet.Extensions.ManagedClient public bool IsConnected => _mqttClient.IsConnected; public bool IsStarted => _connectionCancellationToken != null; + public int PendingApplicationMessagesCount => _messageQueue.Count; public event EventHandler Connected; public event EventHandler Disconnected; + public event EventHandler ApplicationMessageReceived; public event EventHandler ApplicationMessageProcessed; - public event EventHandler SynchronizingSubscriptionsFailed; + + public event EventHandler ConnectingFailed; + public event EventHandler SynchronizingSubscriptionsFailed; public async Task StartAsync(IManagedMqttClientOptions options) { @@ -120,10 +124,13 @@ namespace MQTTnet.Extensions.ManagedClient { if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - foreach (var topicFilter in topicFilters) + lock (_subscriptions) { - _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel; - _subscriptionsNotPushed = true; + foreach (var topicFilter in topicFilters) + { + _subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel; + _subscriptionsNotPushed = true; + } } return Task.FromResult(0); @@ -131,12 +138,17 @@ namespace MQTTnet.Extensions.ManagedClient public Task UnsubscribeAsync(IEnumerable topics) { - foreach (var topic in topics) + if (topics == null) throw new ArgumentNullException(nameof(topics)); + + lock (_subscriptions) { - if (_subscriptions.TryRemove(topic, out _)) + foreach (var topic in topics) { - _unsubscriptions.Add(topic); - _subscriptionsNotPushed = true; + if (_subscriptions.Remove(topic)) + { + _unsubscriptions.Add(topic); + _subscriptionsNotPushed = true; + } } } @@ -210,7 +222,7 @@ namespace MQTTnet.Extensions.ManagedClient } } - private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken) + private void PublishQueuedMessages(CancellationToken cancellationToken) { try { @@ -222,12 +234,9 @@ namespace MQTTnet.Extensions.ManagedClient continue; } - if (cancellationToken.IsCancellationRequested) - { - continue; - } + cancellationToken.ThrowIfCancellationRequested(); - await TryPublishQueuedMessageAsync(message).ConfigureAwait(false); + TryPublishQueuedMessage(message); } } catch (OperationCanceledException) @@ -243,17 +252,13 @@ namespace MQTTnet.Extensions.ManagedClient } } - private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message) + private void TryPublishQueuedMessage(ManagedMqttApplicationMessage message) { Exception transmitException = null; try { - await _mqttClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false); - - if (_storageManager != null) - { - await _storageManager.RemoveAsync(message).ConfigureAwait(false); - } + _mqttClient.PublishAsync(message.ApplicationMessage).GetAwaiter().GetResult(); + _storageManager?.RemoveAsync(message).GetAwaiter().GetResult(); } catch (MqttCommunicationException exception) { @@ -282,13 +287,13 @@ namespace MQTTnet.Extensions.ManagedClient _logger.Info(nameof(ManagedMqttClient), "Synchronizing subscriptions"); List subscriptions; - List unsubscriptions; + HashSet unsubscriptions; lock (_subscriptions) { subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList(); - unsubscriptions = new List(_unsubscriptions); + unsubscriptions = new HashSet(_unsubscriptions); _unsubscriptions.Clear(); _subscriptionsNotPushed = false; @@ -316,7 +321,7 @@ namespace MQTTnet.Extensions.ManagedClient _logger.Warning(exception, "Synchronizing subscriptions failed."); _subscriptionsNotPushed = true; - SynchronizingSubscriptionsFailed?.Invoke(this, EventArgs.Empty); + SynchronizingSubscriptionsFailed?.Invoke(this, new MqttManagedProcessFailedEventArgs(exception)); } } @@ -332,8 +337,9 @@ namespace MQTTnet.Extensions.ManagedClient await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false); return ReconnectionResult.Reconnected; } - catch (Exception) + catch (Exception exception) { + ConnectingFailed?.Invoke(this, new MqttManagedProcessFailedEventArgs(exception)); return ReconnectionResult.NotConnected; } } @@ -364,9 +370,7 @@ namespace MQTTnet.Extensions.ManagedClient _publishingCancellationToken = cts; -#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed - Task.Run(() => PublishQueuedMessagesAsync(cts.Token), cts.Token); -#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed + Task.Factory.StartNew(() => PublishQueuedMessages(cts.Token), cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); } private void StopPublishing() diff --git a/Source/MQTTnet.Extensions.ManagedClient/MqttManagedProcessFailedEventArgs.cs b/Source/MQTTnet.Extensions.ManagedClient/MqttManagedProcessFailedEventArgs.cs new file mode 100644 index 00000000..68dc1608 --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/MqttManagedProcessFailedEventArgs.cs @@ -0,0 +1,14 @@ +using System; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class MqttManagedProcessFailedEventArgs : EventArgs + { + public MqttManagedProcessFailedEventArgs(Exception exception) + { + Exception = exception; + } + + public Exception Exception { get; } + } +}