From 68f1043ef597f09ee718048aa56deca7a537c589 Mon Sep 17 00:00:00 2001 From: Christian Date: Sun, 19 Sep 2021 12:59:32 +0200 Subject: [PATCH] Refactor sub/unsub management in managed client. --- Build/MQTTnet.AspNetCore.nuspec | 2 +- Build/MQTTnet.Extensions.ManagedClient.nuspec | 2 +- Build/MQTTnet.Extensions.Rpc.nuspec | 2 +- Build/MQTTnet.Extensions.WebSocket4Net.nuspec | 2 +- Build/MQTTnet.nuspec | 5 +- .../IManagedMqttClientOptions.cs | 6 +- .../ManagedMqttClient.cs | 77 +++++++++++-------- .../ManagedMqttClientOptions.cs | 4 +- .../ManagedMqttClientOptionsBuilder.cs | 4 +- 9 files changed, 61 insertions(+), 43 deletions(-) diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index f9bc8a04..68fa9dbd 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -12,7 +12,7 @@ true This is a support library to integrate MQTTnet into AspNetCore. For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). - Copyright Christian Kratky 2016-2020 + Copyright Christian Kratky 2016-2021 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 Blazor diff --git a/Build/MQTTnet.Extensions.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index b360569a..8c43ed63 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -12,7 +12,7 @@ true This is an extension library which provides a managed MQTT client with additional features using MQTTnet. For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). - Copyright Christian Kratky 2016-2020 + Copyright Christian Kratky 2016-2021 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 Blazor diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index 283ac11e..d0384138 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -12,7 +12,7 @@ true This is an extension library which allows executing synchronous device calls including a response using MQTTnet. For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). - Copyright Christian Kratky 2016-2020 + Copyright Christian Kratky 2016-2021 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 Blazor diff --git a/Build/MQTTnet.Extensions.WebSocket4Net.nuspec b/Build/MQTTnet.Extensions.WebSocket4Net.nuspec index 640b6909..21baa845 100644 --- a/Build/MQTTnet.Extensions.WebSocket4Net.nuspec +++ b/Build/MQTTnet.Extensions.WebSocket4Net.nuspec @@ -12,7 +12,7 @@ true This is an extension library which allows using _WebSocket4Net_ as transport for MQTTnet clients. For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/). - Copyright Christian Kratky 2016-2020 + Copyright Christian Kratky 2016-2021 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 Blazor diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 5e0bce99..890e8d44 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -12,7 +12,8 @@ true MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol. -* [ManagedClient] Extended ReconnectAsync (thanks to @nvsnkv, #1202). +* [ManagedClient] Extended ReconnectAsync (thanks to @nvsnkv, #1202). +* [ManagedClient] Improved Amazon AWS support (thanks to @scottbrogden-iheartmedia, #1209). * [Server] Fixed a memory/performance leak when using QoS Level 1. * [Server] Exposed connection timestamp in client status. * [Server] Refactored connection management code. @@ -21,7 +22,7 @@ * [MQTTnet, MQTTnet.Extensions.ManagedClient] Fixed bug that allowed invalid subscriptions (Thanks to @marcelwinh). Git commit: $gitCommit - Copyright Christian Kratky 2016-2020 + Copyright Christian Kratky 2016-2021 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 Blazor diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs index cd869709..74028dae 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs @@ -18,6 +18,10 @@ namespace MQTTnet.Extensions.ManagedClient MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; } - int? MaxSubcribeUnsubscribeMessagesAtOnce { get; } + /// + /// Defines the maximum amount of topic filters which will be sent in a SUBSCRIBE/UNSUBSCRIBE packet. + /// Amazon AWS limits this number to 8. The default is int.MaxValue. + /// + int MaxTopicFiltersInSubscribeUnsubscribePackets { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 1c2e7f14..a9e9481f 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -28,6 +28,7 @@ namespace MQTTnet.Extensions.ManagedClient /// at reconnect and are solely owned by . /// readonly Dictionary _reconnectSubscriptions = new Dictionary(); + readonly Dictionary _subscriptions = new Dictionary(); readonly HashSet _unsubscriptions = new HashSet(); readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0); @@ -205,7 +206,6 @@ namespace MQTTnet.Extensions.ManagedClient await applicationMessageSkippedHandler.HandleApplicationMessageSkippedAsync(applicationMessageSkippedEventArgs).ConfigureAwait(false); } } - } } @@ -248,6 +248,7 @@ namespace MQTTnet.Extensions.ManagedClient _unsubscriptions.Add(topic); } } + _subscriptionsQueuedSignal.Release(); return Task.FromResult(0); @@ -481,8 +482,14 @@ namespace MQTTnet.Extensions.ManagedClient lock (_subscriptions) { - subscriptions = _subscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }).ToList(); + subscriptions = _subscriptions.Select(i => new MqttTopicFilter + { + Topic = i.Key, + QualityOfServiceLevel = i.Value + }).ToList(); + _subscriptions.Clear(); + unsubscriptions = new HashSet(_unsubscriptions); _unsubscriptions.Clear(); } @@ -492,7 +499,7 @@ namespace MQTTnet.Extensions.ManagedClient continue; } - _logger.Verbose("Publishing {0} subscriptions and {1} unsubscriptions)", subscriptions.Count, unsubscriptions.Count); + _logger.Verbose("Publishing {0} added and {1} removed subscriptions", subscriptions.Count, unsubscriptions.Count); foreach (var unsubscription in unsubscriptions) { @@ -504,48 +511,48 @@ namespace MQTTnet.Extensions.ManagedClient _reconnectSubscriptions[subscription.Topic] = subscription.QualityOfServiceLevel; } - List subs = new List(); - foreach (var sub in subscriptions) + var addedTopicFilters = new List(); + foreach (var subscription in subscriptions) { - subs.Add(sub); - //aws only allows 8 in a single message - if (subs.Count == Options.MaxSubcribeUnsubscribeMessagesAtOnce) + addedTopicFilters.Add(subscription); + + if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - await SendSubscribeUnsubscribe(subs, null).ConfigureAwait(false); - subs.Clear(); + await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); + addedTopicFilters.Clear(); } } - await SendSubscribeUnsubscribe(subs, null); + await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); - List unSubs = new List(); + var removedTopicFilters = new List(); foreach (var unSub in unsubscriptions) { - unSubs.Add(unSub); - //aws only allows 8 in a single message - if (unSubs.Count == Options.MaxSubcribeUnsubscribeMessagesAtOnce) + removedTopicFilters.Add(unSub); + + if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - await SendSubscribeUnsubscribe(null, unSubs).ConfigureAwait(false); - unSubs.Clear(); + await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); + removedTopicFilters.Clear(); } } - await SendSubscribeUnsubscribe(null, unSubs); + await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); } } - async Task SendSubscribeUnsubscribe(List subscriptions, List unsubscriptions) + async Task SendSubscribeUnsubscribe(List addedSubscriptions, List removedSubscriptions) { try { - if (unsubscriptions != null && unsubscriptions.Any()) + if (removedSubscriptions != null && removedSubscriptions.Any()) { - await InternalClient.UnsubscribeAsync(unsubscriptions.ToArray()).ConfigureAwait(false); + await InternalClient.UnsubscribeAsync(removedSubscriptions.ToArray()).ConfigureAwait(false); } - if (subscriptions != null && subscriptions.Any()) + if (addedSubscriptions != null && addedSubscriptions.Any()) { - await InternalClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false); + await InternalClient.SubscribeAsync(addedSubscriptions.ToArray()).ConfigureAwait(false); } } catch (Exception exception) @@ -562,20 +569,26 @@ namespace MQTTnet.Extensions.ManagedClient { if (_reconnectSubscriptions.Any()) { - var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }); - List subs = new List(); + var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter + { + Topic = i.Key, + QualityOfServiceLevel = i.Value + }); + + var topicFilters = new List(); + foreach (var sub in subscriptions) { - subs.Add(sub); - //aws only allows 8 in a single message - if (subs.Count == Options.MaxSubcribeUnsubscribeMessagesAtOnce) + topicFilters.Add(sub); + + if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - await SendSubscribeUnsubscribe(subs, null).ConfigureAwait(false); - subs.Clear(); + await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); + topicFilters.Clear(); } } - await SendSubscribeUnsubscribe(subs, null).ConfigureAwait(false); + await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); } } catch (Exception exception) @@ -665,4 +678,4 @@ namespace MQTTnet.Extensions.ManagedClient return remainingTime < TimeSpan.Zero ? TimeSpan.Zero : remainingTime; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs index ae5f4d10..f948055e 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs @@ -4,7 +4,7 @@ using MQTTnet.Server; namespace MQTTnet.Extensions.ManagedClient { - public class ManagedMqttClientOptions : IManagedMqttClientOptions + public sealed class ManagedMqttClientOptions : IManagedMqttClientOptions { public IMqttClientOptions ClientOptions { get; set; } @@ -18,6 +18,6 @@ namespace MQTTnet.Extensions.ManagedClient public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropNewMessage; - public int? MaxSubcribeUnsubscribeMessagesAtOnce { get; set; } = null; + public int MaxTopicFiltersInSubscribeUnsubscribePackets { get; set; } = int.MaxValue; } } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs index 9d0ae876..dec105ff 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs @@ -69,9 +69,9 @@ namespace MQTTnet.Extensions.ManagedClient return this; } - public ManagedMqttClientOptionsBuilder WithMaxSubcribeUnsubscribeMessagesAtOnce(int? value) + public ManagedMqttClientOptionsBuilder WithMaxTopicFiltersInSubscribeUnsubscribePackets(int value) { - _options.MaxSubcribeUnsubscribeMessagesAtOnce = value; + _options.MaxTopicFiltersInSubscribeUnsubscribePackets = value; return this; }