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

1862 Add subscribe and unsubscribe results event for Managed Mqtt Client. (#1894)

Browse Source 
* See the issue #1862

Adding an event when ManagedMqttClient actually subscribe/unsubscribe to topics.
This event has to be used in parallel with connected event to know the exact state.

* Rename SubscriptionsResultAsync to SubscriptionsChangedAsync.

* Rename internal field for event

* Update ReleaseNotes.md

---------

Co-authored-by: Pierre Dufrene <pierre.dufrene@se.com>
Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
This commit is contained in:
pdufrene
2024-03-02 09:58:42 +01:00
committed by GitHub
co-authored by Pierre Dufrene Christian
parent 1245d8299a
commit aabf180b23
5 changed files with 155 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ReleaseNotes.md
+1
View File
@@ -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).
Samples/ManagedClient/Managed_Client_Subscribe_Samples.cs
+87
View File
@@ -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;
}
}
Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs
+2
View File
@@ -25,6 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient
event Func<ManagedProcessFailedEventArgs, Task> SynchronizingSubscriptionsFailedAsync;
event Func<SubscriptionsChangedEventArgs, Task> SubscriptionsChangedAsync;
IMqttClient InternalClient { get; }
bool IsConnected { get; }
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+42 -11
View File
@@ -25,7 +25,8 @@ namespace MQTTnet.Extensions.ManagedClient
readonly AsyncEvent<ConnectingFailedEventArgs> _connectingFailedEvent = new AsyncEvent<ConnectingFailedEventArgs>();
readonly AsyncEvent<EventArgs> _connectionStateChangedEvent = new AsyncEvent<EventArgs>();
readonly AsyncEvent<ManagedProcessFailedEventArgs> _synchronizingSubscriptionsFailedEvent = new AsyncEvent<ManagedProcessFailedEventArgs>();
readonly AsyncEvent<SubscriptionsChangedEventArgs> _subscriptionsChangedEvent = new AsyncEvent<SubscriptionsChangedEventArgs>();
readonly MqttNetSourceLogger _logger;
readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
@@ -133,6 +134,12 @@ namespace MQTTnet.Extensions.ManagedClient
remove => _synchronizingSubscriptionsFailedEvent.RemoveHandler(value);
}
public event Func<SubscriptionsChangedEventArgs, Task> 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<MqttClientSubscribeResult> subscribeResults, List<MqttClientUnsubscribeResult> 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<MqttTopicFilter>();
List<MqttClientSubscribeResult> subscribeResults;
List<MqttClientUnsubscribeResult> 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<MqttTopicFilter> subscriptions;
List<MqttClientSubscribeResult> subscribeResults;
List<MqttClientUnsubscribeResult> unsubscribeResults;
HashSet<string> 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<string>();
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<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
async Task<(List<MqttClientSubscribeResult> SubscribeResults, List<MqttClientUnsubscribeResult> UnsubscribeResults)> SendSubscribeUnsubscribe(List<MqttTopicFilter> addedSubscriptions, List<string> removedSubscriptions)
{
var subscribeResults = new List<MqttClientSubscribeResult>();
var unsubscribeResults = new List<MqttClientUnsubscribeResult>();
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()
Source/MQTTnet.Extensions.ManagedClient/SubscriptionsChangedEventArgs.cs
+23
View File
@@ -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<MqttClientSubscribeResult> subscribeResult, List<MqttClientUnsubscribeResult> unsubscribeResult)
{
SubscribeResult = subscribeResult ?? throw new ArgumentNullException(nameof(subscribeResult));
UnsubscribeResult = unsubscribeResult ?? throw new ArgumentNullException(nameof(unsubscribeResult));
}
public List<MqttClientSubscribeResult> SubscribeResult { get; }
public List<MqttClientUnsubscribeResult> UnsubscribeResult { get; }
}
}