diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 087b683a..5fe443bf 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,2 +1,3 @@ * [Core] Optimized packet serialization of PUBACK and PUBREC packets for protocol version 5.0.0 (#1939, thanks to @Y-Sindo). * [Core] The package inspector is now fully async (#1941). +* [ManagedClient] Added a new event (SubscriptionsChangedAsync) which is fired when a subscription or unsubscription was made (#1894, thanks to @pdufrene). diff --git a/Samples/ManagedClient/Managed_Client_Subscribe_Samples.cs b/Samples/ManagedClient/Managed_Client_Subscribe_Samples.cs new file mode 100644 index 00000000..b62e3173 --- /dev/null +++ b/Samples/ManagedClient/Managed_Client_Subscribe_Samples.cs @@ -0,0 +1,87 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +// ReSharper disable UnusedType.Global +// ReSharper disable UnusedMember.Global +// ReSharper disable InconsistentNaming + +using MQTTnet.Client; +using MQTTnet.Extensions.ManagedClient; + +namespace MQTTnet.Samples.ManagedClient; + +public sealed class Managed_Client_Subscribe_Samples +{ + public static async Task Connect_Client() + { + /* + * This sample creates a simple managed MQTT client and connects to a public broker, subscribe to a topic and verifies subscription result. + * + * The managed client extends the existing _MqttClient_. It adds the following features. + * - Reconnecting when connection is lost. + * - Storing pending messages in an internal queue so that an enqueue is possible while the client remains not connected. + */ + + var mqttFactory = new MqttFactory(); + var subscribed = false; + + using (var managedMqttClient = mqttFactory.CreateManagedMqttClient()) + { + var mqttClientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("broker.hivemq.com") + .Build(); + + var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder() + .WithClientOptions(mqttClientOptions) + .Build(); + + await managedMqttClient.StartAsync(managedMqttClientOptions); + + // The application message is not sent. It is stored in an internal queue and + // will be sent when the client is connected. + await managedMqttClient.EnqueueAsync("Topic", "Payload"); + + Console.WriteLine("The managed MQTT client is connected."); + + // Wait until the queue is fully processed. + SpinWait.SpinUntil(() => managedMqttClient.PendingApplicationMessagesCount == 0, 10000); + + Console.WriteLine($"Pending messages = {managedMqttClient.PendingApplicationMessagesCount}"); + + managedMqttClient.SubscriptionsChangedAsync += args => SubscriptionsResultAsync(args, subscribed); + await managedMqttClient.SubscribeAsync("Topic").ConfigureAwait(false); + + SpinWait.SpinUntil(() => subscribed, 1000); + Console.WriteLine("Subscription properly done"); + } + } + + private static Task SubscriptionsResultAsync(SubscriptionsChangedEventArgs arg, bool subscribed) + { + foreach (var mqttClientSubscribeResult in arg.SubscribeResult) + { + Console.WriteLine($"Subscription reason {mqttClientSubscribeResult.ReasonString}"); + foreach (var item in mqttClientSubscribeResult.Items) + { + Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}"); + + if (item.TopicFilter.Topic == "Topic" && item.ResultCode == MqttClientSubscribeResultCode.GrantedQoS0 && !subscribed) + { + subscribed = true; + } + } + } + + foreach (var mqttClientUnsubscribeResult in arg.UnsubscribeResult) + { + Console.WriteLine($"Unsubscription reason {mqttClientUnsubscribeResult.ReasonString}"); + foreach (var item in mqttClientUnsubscribeResult.Items) + { + Console.WriteLine($"For topic filter {item.TopicFilter}, result code: {item.ResultCode}"); + } + } + + return Task.CompletedTask; + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index 49add6a3..f814437f 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -25,6 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient event Func SynchronizingSubscriptionsFailedAsync; + event Func SubscriptionsChangedAsync; + IMqttClient InternalClient { get; } bool IsConnected { get; } diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 32a93807..625bd4c6 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -25,7 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient readonly AsyncEvent _connectingFailedEvent = new AsyncEvent(); readonly AsyncEvent _connectionStateChangedEvent = new AsyncEvent(); readonly AsyncEvent _synchronizingSubscriptionsFailedEvent = new AsyncEvent(); - + readonly AsyncEvent _subscriptionsChangedEvent = new AsyncEvent(); + readonly MqttNetSourceLogger _logger; readonly BlockingQueue _messageQueue = new BlockingQueue(); @@ -133,6 +134,12 @@ namespace MQTTnet.Extensions.ManagedClient remove => _synchronizingSubscriptionsFailedEvent.RemoveHandler(value); } + public event Func SubscriptionsChangedAsync + { + add => _subscriptionsChangedEvent.AddHandler(value); + remove => _subscriptionsChangedEvent.RemoveHandler(value); + } + public IMqttClient InternalClient { get; } public bool IsConnected => InternalClient.IsConnected; @@ -372,6 +379,14 @@ namespace MQTTnet.Extensions.ManagedClient } } + async Task HandleSubscriptionsResultAsync(List subscribeResults, List unsubscribeResults) + { + if (_subscriptionsChangedEvent.HasHandlers) + { + await _subscriptionsChangedEvent.InvokeAsync(new SubscriptionsChangedEventArgs(subscribeResults, unsubscribeResults)).ConfigureAwait(false); + } + } + async Task MaintainConnectionAsync(CancellationToken cancellationToken) { try @@ -473,19 +488,23 @@ namespace MQTTnet.Extensions.ManagedClient if (_reconnectSubscriptions.Any()) { topicFilters = new List(); - + List subscribeResults; + List unsubscribeResults; + foreach (var sub in _reconnectSubscriptions) { topicFilters.Add(sub.Value); if (topicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); + (subscribeResults, unsubscribeResults) = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); topicFilters.Clear(); + await HandleSubscriptionsResultAsync(subscribeResults, unsubscribeResults).ConfigureAwait(false); } } - await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); + (subscribeResults, unsubscribeResults) = await SendSubscribeUnsubscribe(topicFilters, null).ConfigureAwait(false); + await HandleSubscriptionsResultAsync(subscribeResults, unsubscribeResults).ConfigureAwait(false); } } catch (Exception exception) @@ -501,6 +520,8 @@ namespace MQTTnet.Extensions.ManagedClient while (await _subscriptionsQueuedSignal.WaitAsync(GetRemainingTime(endTime), cancellationToken).ConfigureAwait(false)) { List subscriptions; + List subscribeResults; + List unsubscribeResults; HashSet unsubscriptions; lock (_subscriptions) @@ -536,12 +557,14 @@ namespace MQTTnet.Extensions.ManagedClient if (addedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); + (subscribeResults, unsubscribeResults) = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); addedTopicFilters.Clear(); + await HandleSubscriptionsResultAsync(subscribeResults, unsubscribeResults).ConfigureAwait(false); } } - await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); + (subscribeResults, unsubscribeResults) = await SendSubscribeUnsubscribe(addedTopicFilters, null).ConfigureAwait(false); + await HandleSubscriptionsResultAsync(subscribeResults, unsubscribeResults).ConfigureAwait(false); var removedTopicFilters = new List(); foreach (var unSub in unsubscriptions) @@ -550,12 +573,14 @@ namespace MQTTnet.Extensions.ManagedClient if (removedTopicFilters.Count == Options.MaxTopicFiltersInSubscribeUnsubscribePackets) { - await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); + (subscribeResults, unsubscribeResults) = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); removedTopicFilters.Clear(); + await HandleSubscriptionsResultAsync(subscribeResults, unsubscribeResults).ConfigureAwait(false); } } - await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); + (subscribeResults, unsubscribeResults) = await SendSubscribeUnsubscribe(null, removedTopicFilters).ConfigureAwait(false); + await HandleSubscriptionsResultAsync(subscribeResults, unsubscribeResults).ConfigureAwait(false); } } @@ -588,8 +613,10 @@ namespace MQTTnet.Extensions.ManagedClient } } - async Task SendSubscribeUnsubscribe(List addedSubscriptions, List removedSubscriptions) + async Task<(List SubscribeResults, List UnsubscribeResults)> SendSubscribeUnsubscribe(List addedSubscriptions, List removedSubscriptions) { + var subscribeResults = new List(); + var unsubscribeResults = new List(); try { if (removedSubscriptions != null && removedSubscriptions.Any()) @@ -601,7 +628,8 @@ namespace MQTTnet.Extensions.ManagedClient unsubscribeOptionsBuilder.WithTopicFilter(removedSubscription); } - await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false); + var unsubscribeResult = await InternalClient.UnsubscribeAsync(unsubscribeOptionsBuilder.Build()).ConfigureAwait(false); + unsubscribeResults.Add(unsubscribeResult); //clear because these worked, maybe the subscribe below will fail, only report those removedSubscriptions.Clear(); @@ -616,13 +644,16 @@ namespace MQTTnet.Extensions.ManagedClient subscribeOptionsBuilder.WithTopicFilter(addedSubscription); } - await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false); + var subscribeResult = await InternalClient.SubscribeAsync(subscribeOptionsBuilder.Build()).ConfigureAwait(false); + subscribeResults.Add(subscribeResult); } } catch (Exception exception) { await HandleSubscriptionExceptionAsync(exception, addedSubscriptions, removedSubscriptions).ConfigureAwait(false); } + + return (subscribeResults, unsubscribeResults); } void StartPublishing() diff --git a/Source/MQTTnet.Extensions.ManagedClient/SubscriptionsChangedEventArgs.cs b/Source/MQTTnet.Extensions.ManagedClient/SubscriptionsChangedEventArgs.cs new file mode 100644 index 00000000..3118001f --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/SubscriptionsChangedEventArgs.cs @@ -0,0 +1,23 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using MQTTnet.Client; +using System; +using System.Collections.Generic; + +namespace MQTTnet.Extensions.ManagedClient +{ + public sealed class SubscriptionsChangedEventArgs : EventArgs + { + public SubscriptionsChangedEventArgs(List subscribeResult, List unsubscribeResult) + { + SubscribeResult = subscribeResult ?? throw new ArgumentNullException(nameof(subscribeResult)); + UnsubscribeResult = unsubscribeResult ?? throw new ArgumentNullException(nameof(unsubscribeResult)); + } + + public List SubscribeResult { get; } + + public List UnsubscribeResult { get; } + } +}