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

Squashed commit of the following:

Browse Source 
commit 0cbf50427ca6fb343cfa22ccbcd06aad5ea2b4c2
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Mon Jun 20 21:02:50 2022 +0200

    Fix firing of _ClientSubscribedTopicAsync_ event.
This commit is contained in:
Christian
2022-06-27 20:41:34 +02:00
parent f632fbb447
commit 305c552474
4 changed files with 66 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Tests/Server/Subscribe_Tests.cs
+26 -1
View File
@@ -12,6 +12,7 @@ using MQTTnet.Formatter;
using MQTTnet.Implementations;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Server
{
@@ -45,6 +46,30 @@ namespace MQTTnet.Tests.Server
}
}
[TestMethod]
public async Task Enqueue_Message_After_Subscription()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
server.ClientSubscribedTopicAsync += e =>
{
server.InjectApplicationMessage(new InjectedMqttApplicationMessage(new MqttApplicationMessageBuilder().WithTopic("test_topic").Build()));
return PlatformAbstractionLayer.CompletedTask;
};
var client = await testEnvironment.ConnectClient();
var receivedMessages = testEnvironment.CreateApplicationMessageHandler(client);
await client.SubscribeAsync("test_topic");
await LongTestDelay();
Assert.AreEqual(1, receivedMessages.ReceivedEventArgs.Count);
}
}
[TestMethod]
public async Task Intercept_Subscription()
{
@@ -101,7 +126,7 @@ namespace MQTTnet.Tests.Server
.WithTopicFilter("c", MqttQualityOfServiceLevel.ExactlyOnce)
.WithTopicFilter("d")
.Build();
var response = await client.SubscribeAsync(subscribeOptions);
Assert.AreEqual(subscribeOptions.TopicFilters.Count, response.Items.Count);
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+37 -30
View File
@@ -175,13 +175,14 @@ namespace MQTTnet.Server
};
var addedSubscriptions = new List<string>();
var finalTopicFilters = new List<MqttTopicFilter>();
// The topic filters are order by its QoS so that the higher QoS will win over a
// lower one.
foreach (var originalTopicFilter in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel))
foreach (var topicFilterItem in subscribePacket.TopicFilters.OrderByDescending(f => f.QualityOfServiceLevel))
{
var subscriptionEventArgs = await InterceptSubscribe(originalTopicFilter, cancellationToken).ConfigureAwait(false);
var finalTopicFilter = subscriptionEventArgs.TopicFilter;
var subscriptionEventArgs = await InterceptSubscribe(topicFilterItem, cancellationToken).ConfigureAwait(false);
var topicFilter = subscriptionEventArgs.TopicFilter;
var processSubscription = subscriptionEventArgs.ProcessSubscription && subscriptionEventArgs.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
result.UserProperties = subscriptionEventArgs.UserProperties;
@@ -195,16 +196,26 @@ namespace MQTTnet.Server
result.CloseConnection = true;
}
if (!processSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic))
if (!processSubscription || string.IsNullOrEmpty(topicFilter.Topic))
{
continue;
}
var createSubscriptionResult = CreateSubscription(finalTopicFilter, subscribePacket.SubscriptionIdentifier, subscriptionEventArgs.Response.ReasonCode);
var createSubscriptionResult = CreateSubscription(topicFilter, subscribePacket.SubscriptionIdentifier, subscriptionEventArgs.Response.ReasonCode);
addedSubscriptions.Add(finalTopicFilter.Topic);
addedSubscriptions.Add(topicFilter.Topic);
finalTopicFilters.Add(topicFilter);
if (_eventContainer.ClientSubscribedTopicEvent.HasHandlers)
FilterRetainedApplicationMessages(retainedApplicationMessages, createSubscriptionResult, result);
}
// This call will add the new subscription to the internal storage.
// So the event _ClientSubscribedTopicEvent_ must be called afterwards.
_subscriptionChangedNotification?.OnSubscriptionsAdded(_session, addedSubscriptions);
if (_eventContainer.ClientSubscribedTopicEvent.HasHandlers)
{
foreach (var finalTopicFilter in finalTopicFilters)
{
var eventArgs = new ClientSubscribedTopicEventArgs
{
@@ -214,12 +225,8 @@ namespace MQTTnet.Server
await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
FilterRetainedApplicationMessages(retainedApplicationMessages, createSubscriptionResult, result);
}
_subscriptionChangedNotification?.OnSubscriptionsAdded(_session, addedSubscriptions);
return result;
}
@@ -263,26 +270,30 @@ namespace MQTTnet.Server
_subscriptions.Remove(topicFilter);
// must remove subscription object from topic hash dictionary also
if (existingSubscription.TopicHasWildcard)
if (existingSubscription != null)
{
if (_wildcardSubscriptionsByTopicHash.TryGetValue(existingSubscription.TopicHash, out var subs))
var topicHash = existingSubscription.TopicHash;
if (existingSubscription.TopicHasWildcard)
{
subs.Subscriptions.Remove(existingSubscription);
if (subs.Subscriptions.Count == 0)
if (_wildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions))
{
_wildcardSubscriptionsByTopicHash.Remove(existingSubscription.TopicHash);
subscriptions.Subscriptions.Remove(existingSubscription);
if (subscriptions.Subscriptions.Count == 0)
{
_wildcardSubscriptionsByTopicHash.Remove(topicHash);
}
}
}
}
else
{
if (_noWildcardSubscriptionsByTopicHash.TryGetValue(existingSubscription.TopicHash, out var subs))
else
{
subs.Remove(existingSubscription);
if (subs.Count == 0)
if (_noWildcardSubscriptionsByTopicHash.TryGetValue(topicHash, out var subscriptions))
{
_noWildcardSubscriptionsByTopicHash.Remove(existingSubscription.TopicHash);
subscriptions.Remove(existingSubscription);
if (subscriptions.Count == 0)
{
_noWildcardSubscriptionsByTopicHash.Remove(topicHash);
}
}
}
}
@@ -294,11 +305,7 @@ namespace MQTTnet.Server
finally
{
_subscriptionsLock.Release();
if (_subscriptionChangedNotification != null)
{
_subscriptionChangedNotification.OnSubscriptionsRemoved(_session, removedSubscriptions);
}
_subscriptionChangedNotification?.OnSubscriptionsRemoved(_session, removedSubscriptions);
}
if (_eventContainer.ClientUnsubscribedTopicEvent.HasHandlers)