diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs index a705efea..cd869709 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs @@ -1,4 +1,4 @@ -using System; +using System; using MQTTnet.Client.Options; using MQTTnet.Server; @@ -17,5 +17,7 @@ namespace MQTTnet.Extensions.ManagedClient int MaxPendingMessages { get; } MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; } + + int? MaxSubcribeUnsubscribeMessagesAtOnce { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index d349f81e..1c2e7f14 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -1,4 +1,4 @@ -using MQTTnet.Client; +using MQTTnet.Client; using MQTTnet.Client.Connecting; using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Publishing; @@ -504,22 +504,53 @@ namespace MQTTnet.Extensions.ManagedClient _reconnectSubscriptions[subscription.Topic] = subscription.QualityOfServiceLevel; } - try + List subs = new List(); + foreach (var sub in subscriptions) { - if (unsubscriptions.Any()) + subs.Add(sub); + //aws only allows 8 in a single message + if (subs.Count == Options.MaxSubcribeUnsubscribeMessagesAtOnce) { - await InternalClient.UnsubscribeAsync(unsubscriptions.ToArray()).ConfigureAwait(false); + await SendSubscribeUnsubscribe(subs, null).ConfigureAwait(false); + subs.Clear(); } + } - if (subscriptions.Any()) + await SendSubscribeUnsubscribe(subs, null); + + List unSubs = new List(); + foreach (var unSub in unsubscriptions) + { + unSubs.Add(unSub); + //aws only allows 8 in a single message + if (unSubs.Count == Options.MaxSubcribeUnsubscribeMessagesAtOnce) { - await InternalClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false); + await SendSubscribeUnsubscribe(null, unSubs).ConfigureAwait(false); + unSubs.Clear(); } } - catch (Exception exception) + + await SendSubscribeUnsubscribe(null, unSubs); + } + } + + async Task SendSubscribeUnsubscribe(List subscriptions, List unsubscriptions) + { + try + { + if (unsubscriptions != null && unsubscriptions.Any()) { - await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false); + await InternalClient.UnsubscribeAsync(unsubscriptions.ToArray()).ConfigureAwait(false); } + + if (subscriptions != null && subscriptions.Any()) + { + await InternalClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false); + } + } + catch (Exception exception) + { + await HandleSubscriptionExceptionAsync(exception).ConfigureAwait(false); } } @@ -532,7 +563,19 @@ namespace MQTTnet.Extensions.ManagedClient if (_reconnectSubscriptions.Any()) { var subscriptions = _reconnectSubscriptions.Select(i => new MqttTopicFilter { Topic = i.Key, QualityOfServiceLevel = i.Value }); - await InternalClient.SubscribeAsync(subscriptions.ToArray()).ConfigureAwait(false); + List subs = new List(); + foreach (var sub in subscriptions) + { + subs.Add(sub); + //aws only allows 8 in a single message + if (subs.Count == Options.MaxSubcribeUnsubscribeMessagesAtOnce) + { + await SendSubscribeUnsubscribe(subs, null).ConfigureAwait(false); + subs.Clear(); + } + } + + await SendSubscribeUnsubscribe(subs, null).ConfigureAwait(false); } } catch (Exception exception) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs index 05035bb3..ae5f4d10 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs @@ -1,4 +1,4 @@ -using System; +using System; using MQTTnet.Client.Options; using MQTTnet.Server; @@ -17,5 +17,7 @@ namespace MQTTnet.Extensions.ManagedClient public int MaxPendingMessages { get; set; } = int.MaxValue; public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropNewMessage; + + public int? MaxSubcribeUnsubscribeMessagesAtOnce { get; set; } = null; } } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs index 6157655b..9d0ae876 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs @@ -1,4 +1,4 @@ -using System; +using System; using MQTTnet.Client.Options; using MQTTnet.Server; @@ -68,7 +68,13 @@ namespace MQTTnet.Extensions.ManagedClient options(_clientOptionsBuilder); return this; } - + + public ManagedMqttClientOptionsBuilder WithMaxSubcribeUnsubscribeMessagesAtOnce(int? value) + { + _options.MaxSubcribeUnsubscribeMessagesAtOnce = value; + return this; + } + public ManagedMqttClientOptions Build() { if (_clientOptionsBuilder != null)