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

Add user properties to event args to interceptors for subscribing and unsubscribing (#1707)

Browse Source 
Add user properties to event args
This commit is contained in:
Christian
2023-04-06 17:31:42 +02:00
committed by GitHub
parent b4f740f338
commit d6cab76728
9 changed files with 99 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ReleaseNotes.md
+1
View File
@@ -12,3 +12,4 @@
* [Server] Added a new API method which allows reading a single retained message without the need to processing the entire set of retained messages (#1659).
* [Server] Added a new event (InterceptingClientEnqueueAsync) which allows intercepting when an application message is enqueued for a client (#1648).
* [Server] Fixed race condition when handling connections which leads to stopped message transfers (#1677, thanks to @RazvanEmilR).
* [Server] Added missing user properties to subscribe and unsubscribe interceptor (#1702).
Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs
+2
View File
@@ -99,6 +99,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
// Perform a clean disconnect.
await client.DisconnectAsync(disconnectOptions);
await LongTestDelay();
Assert.IsNotNull(eventArgs);
Assert.AreEqual(MqttClientDisconnectType.Clean, eventArgs.DisconnectType);
}
Source/MQTTnet.Tests/Server/Subscribe_Tests.cs
+25 -2
View File
@@ -62,7 +62,7 @@ namespace MQTTnet.Tests.Server
}
}
}
[TestMethod]
public async Task Deny_Invalid_Topic()
{
@@ -90,6 +90,29 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
public async Task Intercept_Subscribe_With_User_Properties()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
var server = await testEnvironment.StartServer();
InterceptingSubscriptionEventArgs eventArgs = null;
server.InterceptingSubscriptionAsync += e =>
{
eventArgs = e;
return CompletedTask.Instance;
};
var client = await testEnvironment.ConnectClient();
var subscribeOptions = testEnvironment.Factory.CreateSubscribeOptionsBuilder().WithTopicFilter("X").WithUserProperty("A", "1").Build();
await client.SubscribeAsync(subscribeOptions);
CollectionAssert.AreEqual(subscribeOptions.UserProperties.ToList(), eventArgs.UserProperties);
}
}
[TestMethod]
[ExpectedException(typeof(MqttClientDisconnectedException))]
public async Task Disconnect_While_Subscribing()
@@ -105,7 +128,7 @@ namespace MQTTnet.Tests.Server
await client.SubscribeAsync("#");
}
}
[TestMethod]
public async Task Enqueue_Message_After_Subscription()
{
Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs
+27
View File
@@ -2,10 +2,14 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Linq;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Server
{
@@ -28,5 +32,28 @@ namespace MQTTnet.Tests.Server
await client.UnsubscribeAsync("#");
}
}
[TestMethod]
public async Task Intercept_Unsubscribe_With_User_Properties()
{
using (var testEnvironment = CreateTestEnvironment(MqttProtocolVersion.V500))
{
var server = await testEnvironment.StartServer();
InterceptingUnsubscriptionEventArgs eventArgs = null;
server.InterceptingUnsubscriptionAsync += e =>
{
eventArgs = e;
return CompletedTask.Instance;
};
var client = await testEnvironment.ConnectClient();
var unsubscribeOptions = testEnvironment.Factory.CreateUnsubscribeOptionsBuilder().WithTopicFilter("X").WithUserProperty("A", "1").Build();
await client.UnsubscribeAsync(unsubscribeOptions);
CollectionAssert.AreEqual(unsubscribeOptions.UserProperties.ToList(), eventArgs.UserProperties);
}
}
}
}
Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs
+1 -1
View File
@@ -42,6 +42,6 @@ namespace MQTTnet.Client
/// Gets the user properties which were part of the UNSUBACK packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; }
public IReadOnlyCollection<MqttUserProperty> UserProperties { get; set; }
}
}
Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs
+4 -1
View File
@@ -16,12 +16,14 @@ namespace MQTTnet.Server
CancellationToken cancellationToken,
string clientId,
MqttSessionStatus session,
MqttTopicFilter topicFilter)
MqttTopicFilter topicFilter,
List<MqttUserProperty> userProperties)
{
CancellationToken = cancellationToken;
ClientId = clientId;
Session = session;
TopicFilter = topicFilter;
UserProperties = userProperties;
}
/// <summary>
@@ -75,6 +77,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets or sets the user properties.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties { get; set; }
}
Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs
+10 -1
View File
@@ -4,18 +4,21 @@
using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class InterceptingUnsubscriptionEventArgs : EventArgs
{
public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, IDictionary sessionItems, string topic)
public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, IDictionary sessionItems, string topic, List<MqttUserProperty> userProperties)
{
CancellationToken = cancellationToken;
ClientId = clientId;
SessionItems = sessionItems;
Topic = topic;
UserProperties = userProperties;
}
/// <summary>
@@ -59,5 +62,11 @@ namespace MQTTnet.Server
/// separator).
/// </summary>
public string Topic { get; }
/// <summary>
/// Gets or sets the user properties.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties { get; set; }
}
}
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+25 -17
View File
@@ -177,15 +177,15 @@ namespace MQTTnet.Server
// lower one.
foreach (var topicFilterItem in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel))
{
var subscriptionEventArgs = await InterceptSubscribe(topicFilterItem, cancellationToken).ConfigureAwait(false);
var topicFilter = subscriptionEventArgs.TopicFilter;
var processSubscription = subscriptionEventArgs.ProcessSubscription && subscriptionEventArgs.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
var interceptorEventArgs = await InterceptSubscribe(topicFilterItem, subscribePacket.UserProperties, cancellationToken).ConfigureAwait(false);
var topicFilter = interceptorEventArgs.TopicFilter;
var processSubscription = interceptorEventArgs.ProcessSubscription && interceptorEventArgs.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
result.UserProperties = subscriptionEventArgs.UserProperties;
result.ReasonString = subscriptionEventArgs.ReasonString;
result.ReasonCodes.Add(subscriptionEventArgs.Response.ReasonCode);
result.UserProperties = interceptorEventArgs.UserProperties;
result.ReasonString = interceptorEventArgs.ReasonString;
result.ReasonCodes.Add(interceptorEventArgs.Response.ReasonCode);
if (subscriptionEventArgs.CloseConnection)
if (interceptorEventArgs.CloseConnection)
{
// When any of the interceptor calls leads to a connection close the connection
// must be closed. So do not revert to false!
@@ -197,7 +197,7 @@ namespace MQTTnet.Server
continue;
}
var createSubscriptionResult = CreateSubscription(topicFilter, subscribePacket.SubscriptionIdentifier, subscriptionEventArgs.Response.ReasonCode);
var createSubscriptionResult = CreateSubscription(topicFilter, subscribePacket.SubscriptionIdentifier, interceptorEventArgs.Response.ReasonCode);
addedSubscriptions.Add(topicFilter.Topic);
finalTopicFilters.Add(topicFilter);
@@ -238,12 +238,13 @@ namespace MQTTnet.Server
{
_subscriptions.TryGetValue(topicFilter, out var existingSubscription);
var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription, cancellationToken).ConfigureAwait(false);
var acceptUnsubscription = interceptorContext.Response.ReasonCode == MqttUnsubscribeReasonCode.Success;
var interceptorEventArgs = await InterceptUnsubscribe(topicFilter, existingSubscription, unsubscribePacket.UserProperties, cancellationToken).ConfigureAwait(false);
var acceptUnsubscription = interceptorEventArgs.Response.ReasonCode == MqttUnsubscribeReasonCode.Success;
result.ReasonCodes.Add(interceptorContext.Response.ReasonCode);
result.UserProperties = interceptorEventArgs.UserProperties;
result.ReasonCodes.Add(interceptorEventArgs.Response.ReasonCode);
if (interceptorContext.CloseConnection)
if (interceptorEventArgs.CloseConnection)
{
// When any of the interceptor calls leads to a connection close the connection
// must be closed. So do not revert to false!
@@ -255,7 +256,7 @@ namespace MQTTnet.Server
continue;
}
if (interceptorContext.ProcessUnsubscription)
if (interceptorEventArgs.ProcessUnsubscription)
{
_subscriptions.Remove(topicFilter);
@@ -452,9 +453,12 @@ namespace MQTTnet.Server
}
}
async Task<InterceptingSubscriptionEventArgs> InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
async Task<InterceptingSubscriptionEventArgs> InterceptSubscribe(
MqttTopicFilter topicFilter,
List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{
var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, new MqttSessionStatus(_session), topicFilter);
var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, new MqttSessionStatus(_session), topicFilter, userProperties);
if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{
@@ -481,9 +485,13 @@ namespace MQTTnet.Server
return eventArgs;
}
async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(
string topicFilter,
MqttSubscription mqttSubscription,
List<MqttUserProperty> userProperties,
CancellationToken cancellationToken)
{
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, _session.Id, _session.Items, topicFilter)
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, _session.Id, _session.Items, topicFilter, userProperties)
{
Response =
{
Source/MQTTnet/Server/Internal/UnsubscribeResult.cs
+4 -1
View File
@@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server
@@ -10,7 +11,9 @@ namespace MQTTnet.Server
public sealed class UnsubscribeResult
{
public List<MqttUnsubscribeReasonCode> ReasonCodes { get; } = new List<MqttUnsubscribeReasonCode>(128);
public bool CloseConnection { get; set; }
public List<MqttUserProperty> UserProperties { get; set; }
}
}