diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 91530ad9..ceb54560 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,3 +1,4 @@ +* [Core] Updated nuget packages. * [Server] Fixed reporting of _MaximumQoS_ in _ConnAck_ packet (MQTTv5 only) (#1442). * [Server] Fix cross thread issue in session message storage for QoS 1 and 2. -* [Core] Updated nuget packages. +* [Server] The event _ClientSubscribedTopicAsync_ is now fired after the subscription is completely processed internally (#1435). \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 548826f6..359fb294 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -3,7 +3,7 @@ name: CI on: [push, pull_request] env: - VERSION: "4.0.1.${{github.run_number}}" + VERSION: "4.0.2.${{github.run_number}}" jobs: build: diff --git a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs index cc263fe1..862c2202 100644 --- a/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Subscribe_Tests.cs @@ -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); diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index b1174fee..9f2f4c71 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -175,13 +175,14 @@ namespace MQTTnet.Server }; var addedSubscriptions = new List(); + var finalTopicFilters = new List(); // 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)