diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md
index d534f1b7..22322c50 100644
--- a/.github/workflows/ReleaseNotes.md
+++ b/.github/workflows/ReleaseNotes.md
@@ -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).
\ No newline at end of file
diff --git a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs
index 87ecf0b4..1cdcd760 100644
--- a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs
+++ b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Connection_Tests.cs
@@ -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);
}
diff --git a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs
index a4ce93b7..897c3c07 100644
--- a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs
+++ b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs
@@ -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()
{
diff --git a/Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs b/Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs
index 5692fa8d..4ed527c8 100644
--- a/Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs
+++ b/Source/MQTTnet.Tests/Server/Unsubscribe_Tests.cs
@@ -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);
+ }
+ }
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs
index 24b8008f..60aedcdf 100644
--- a/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs
+++ b/Source/MQTTnet/Client/Unsubscribing/MqttClientUnsubscribeResult.cs
@@ -42,6 +42,6 @@ namespace MQTTnet.Client
/// Gets the user properties which were part of the UNSUBACK packet.
/// MQTT 5.0.0+ feature.
///
- public IReadOnlyCollection UserProperties { get; }
+ public IReadOnlyCollection UserProperties { get; set; }
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs
index d7f787e1..fa26d13e 100644
--- a/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs
+++ b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs
@@ -16,12 +16,14 @@ namespace MQTTnet.Server
CancellationToken cancellationToken,
string clientId,
MqttSessionStatus session,
- MqttTopicFilter topicFilter)
+ MqttTopicFilter topicFilter,
+ List userProperties)
{
CancellationToken = cancellationToken;
ClientId = clientId;
Session = session;
TopicFilter = topicFilter;
+ UserProperties = userProperties;
}
///
@@ -75,6 +77,7 @@ namespace MQTTnet.Server
///
/// Gets or sets the user properties.
+ /// MQTT 5.0.0+ feature.
///
public List UserProperties { get; set; }
}
diff --git a/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs
index 63e8cad5..36d08df3 100644
--- a/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs
+++ b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs
@@ -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 userProperties)
{
CancellationToken = cancellationToken;
ClientId = clientId;
SessionItems = sessionItems;
Topic = topic;
+ UserProperties = userProperties;
}
///
@@ -59,5 +62,11 @@ namespace MQTTnet.Server
/// separator).
///
public string Topic { get; }
+
+ ///
+ /// Gets or sets the user properties.
+ /// MQTT 5.0.0+ feature.
+ ///
+ public List UserProperties { get; set; }
}
}
\ No newline at end of file
diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
index af2d8dd3..24595fd4 100644
--- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
@@ -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 InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
+ async Task InterceptSubscribe(
+ MqttTopicFilter topicFilter,
+ List 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 InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
+ async Task InterceptUnsubscribe(
+ string topicFilter,
+ MqttSubscription mqttSubscription,
+ List 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 =
{
diff --git a/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs b/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs
index 5a93d997..38033b19 100644
--- a/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs
+++ b/Source/MQTTnet/Server/Internal/UnsubscribeResult.cs
@@ -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 ReasonCodes { get; } = new List(128);
-
+
public bool CloseConnection { get; set; }
+
+ public List UserProperties { get; set; }
}
}
\ No newline at end of file