From 75dfe68a948eda55a89065efa77aadc6b263b88a Mon Sep 17 00:00:00 2001 From: SeppPenner Date: Tue, 9 Jun 2020 20:54:42 +0200 Subject: [PATCH 1/6] Added interceptor for undelivered messages. --- Build/MQTTnet.nuspec | 3 +- Source/MQTTnet/MqttApplicationMessage.cs | 2 ++ Source/MQTTnet/Server/IMqttServerOptions.cs | 2 ++ Source/MQTTnet/Server/MqttClientSession.cs | 2 ++ .../Server/MqttClientSessionsManager.cs | 12 ++++++++ Source/MQTTnet/Server/MqttServerOptions.cs | 2 ++ .../Server/MqttServerOptionsBuilder.cs | 6 ++++ Tests/MQTTnet.Core.Tests/Server_Tests.cs | 28 +++++++++++++++++++ 8 files changed, 56 insertions(+), 1 deletion(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index b329846d..b1aa6050 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -16,7 +16,8 @@ * [AspNetCore] Adjusted some namespaces (BREAKING CHANGE!) * [Server] Adjusted some namespaces (BREAKING CHANGE!) * [Server] Added state checks (throw if not started etc.) for most server APIs. -* [Server] Exposed real X509Certificate2 (instead byte array) to TLS options (thanks to @borigas). +* [Server] Exposed real X509Certificate2 (instead byte array) to TLS options (Thanks to @borigas). +* [Core] Added server interceptor for undelivered messages (Thanks to @cshark-inator). Copyright Christian Kratky 2016-2020 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin Blazor diff --git a/Source/MQTTnet/MqttApplicationMessage.cs b/Source/MQTTnet/MqttApplicationMessage.cs index 6522d42f..688bb65b 100644 --- a/Source/MQTTnet/MqttApplicationMessage.cs +++ b/Source/MQTTnet/MqttApplicationMessage.cs @@ -29,5 +29,7 @@ namespace MQTTnet public byte[] CorrelationData { get; set; } public List SubscriptionIdentifiers { get; set; } + + public uint DeliveryCount { get; set; } } } diff --git a/Source/MQTTnet/Server/IMqttServerOptions.cs b/Source/MQTTnet/Server/IMqttServerOptions.cs index 7df6f542..61900edb 100644 --- a/Source/MQTTnet/Server/IMqttServerOptions.cs +++ b/Source/MQTTnet/Server/IMqttServerOptions.cs @@ -25,5 +25,7 @@ namespace MQTTnet.Server IMqttServerStorage Storage { get; } IMqttRetainedMessagesManager RetainedMessagesManager { get; } + + IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index c7d2e50d..1e8a2f6d 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -56,6 +56,8 @@ namespace MQTTnet.Server _logger.Verbose("Queued application message with topic '{0}' (ClientId: {1}).", applicationMessage.Topic, ClientId); + applicationMessage.DeliveryCount++; + ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage); } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 7850780f..8be714d1 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -224,6 +224,8 @@ namespace MQTTnet.Server await _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false); } + applicationMessage.DeliveryCount = 0; + foreach (var clientSession in _sessions.Values) { clientSession.EnqueueApplicationMessage( @@ -231,6 +233,16 @@ namespace MQTTnet.Server sender?.ClientId, false); } + + if (applicationMessage.DeliveryCount == 0) + { + if (_options.UndeliveredMessageInterceptor == null) + { + throw new OperationCanceledException(nameof(_options.UndeliveredMessageInterceptor)); + } + + await _options.UndeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(sender?.ClientId, sender?.Session?.Items, applicationMessage)); + } } catch (OperationCanceledException) { diff --git a/Source/MQTTnet/Server/MqttServerOptions.cs b/Source/MQTTnet/Server/MqttServerOptions.cs index 9773e721..b1c595e3 100644 --- a/Source/MQTTnet/Server/MqttServerOptions.cs +++ b/Source/MQTTnet/Server/MqttServerOptions.cs @@ -31,5 +31,7 @@ namespace MQTTnet.Server public IMqttServerStorage Storage { get; set; } public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; } = new MqttRetainedMessagesManager(); + + public IMqttServerApplicationMessageInterceptor UndeliveredMessageInterceptor { get; set; } } } diff --git a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs index b52215f7..4af56729 100644 --- a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs +++ b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs @@ -212,5 +212,11 @@ namespace MQTTnet.Server { return _options; } + + public MqttServerOptionsBuilder WithUndeliveredMessageInterceptor(Action value) + { + _options.UndeliveredMessageInterceptor = new MqttServerApplicationMessageInterceptorDelegate(value); + return this; + } } } diff --git a/Tests/MQTTnet.Core.Tests/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/Server_Tests.cs index 18c106d7..af1b7153 100644 --- a/Tests/MQTTnet.Core.Tests/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server_Tests.cs @@ -1384,5 +1384,33 @@ namespace MQTTnet.Tests Assert.AreEqual(expectedReceivedMessagesCount, receivedMessagesCount); } } + + [TestMethod] + public async Task Intercept_Undelivered() + { + using (var testEnvironment = new TestEnvironment()) + { + var undeliverd = string.Empty; + var svr = await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder().WithUndeliveredMessageInterceptor( + context => + { + undeliverd = context.ApplicationMessage.Topic; + })); + + var topicAReceived = false; + var topicBReceived = false; + + var client = await testEnvironment.ConnectClientAsync(); + + await client.SubscribeAsync("b"); + + await client.PublishAsync("a", null, MqttQualityOfServiceLevel.ExactlyOnce); + + await Task.Delay(500); + + Assert.AreEqual(undeliverd, "a"); + + } + } } } From c20142671f60a5f86c4a0a21018bee005cff77fe Mon Sep 17 00:00:00 2001 From: SeppPenner Date: Wed, 10 Jun 2020 11:52:55 +0200 Subject: [PATCH 2/6] MQTTnet: Adjusted UndeliveredMessageInterceptor handler to return if it's null instead of throwing an exception. --- Source/MQTTnet/Server/MqttClientSessionsManager.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 8be714d1..456636a2 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -238,7 +238,7 @@ namespace MQTTnet.Server { if (_options.UndeliveredMessageInterceptor == null) { - throw new OperationCanceledException(nameof(_options.UndeliveredMessageInterceptor)); + return; } await _options.UndeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(sender?.ClientId, sender?.Session?.Items, applicationMessage)); From 7cf12dc7511acfe741fbae1ee19b5ef7483ed5e2 Mon Sep 17 00:00:00 2001 From: SeppPenner Date: Wed, 10 Jun 2020 18:45:15 +0200 Subject: [PATCH 3/6] Made delivery count internal. --- Source/MQTTnet/MqttApplicationMessage.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet/MqttApplicationMessage.cs b/Source/MQTTnet/MqttApplicationMessage.cs index 688bb65b..65200740 100644 --- a/Source/MQTTnet/MqttApplicationMessage.cs +++ b/Source/MQTTnet/MqttApplicationMessage.cs @@ -30,6 +30,6 @@ namespace MQTTnet public List SubscriptionIdentifiers { get; set; } - public uint DeliveryCount { get; set; } + internal uint DeliveryCount { get; set; } } } From a407fc381bb9091634a111853863b8bca5fd69d0 Mon Sep 17 00:00:00 2001 From: SeppPenner Date: Thu, 11 Jun 2020 12:39:56 +0200 Subject: [PATCH 4/6] Made deliveryCount a local variable. --- Source/MQTTnet/MqttApplicationMessage.cs | 2 -- Source/MQTTnet/Server/MqttClientSession.cs | 2 -- Source/MQTTnet/Server/MqttClientSessionsManager.cs | 6 ++++-- 3 files changed, 4 insertions(+), 6 deletions(-) diff --git a/Source/MQTTnet/MqttApplicationMessage.cs b/Source/MQTTnet/MqttApplicationMessage.cs index 65200740..6522d42f 100644 --- a/Source/MQTTnet/MqttApplicationMessage.cs +++ b/Source/MQTTnet/MqttApplicationMessage.cs @@ -29,7 +29,5 @@ namespace MQTTnet public byte[] CorrelationData { get; set; } public List SubscriptionIdentifiers { get; set; } - - internal uint DeliveryCount { get; set; } } } diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index 1e8a2f6d..c7d2e50d 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -56,8 +56,6 @@ namespace MQTTnet.Server _logger.Verbose("Queued application message with topic '{0}' (ClientId: {1}).", applicationMessage.Topic, ClientId); - applicationMessage.DeliveryCount++; - ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage); } diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index 456636a2..a0f0bf2e 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -224,7 +224,7 @@ namespace MQTTnet.Server await _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false); } - applicationMessage.DeliveryCount = 0; + var deliveryCount = 0; foreach (var clientSession in _sessions.Values) { @@ -232,9 +232,11 @@ namespace MQTTnet.Server applicationMessage, sender?.ClientId, false); + + deliveryCount++; } - if (applicationMessage.DeliveryCount == 0) + if (deliveryCount == 0) { if (_options.UndeliveredMessageInterceptor == null) { From 257211fc7a98d62895a6f16f673de7d9a17c5f4a Mon Sep 17 00:00:00 2001 From: SeppPenner Date: Thu, 11 Jun 2020 12:41:48 +0200 Subject: [PATCH 5/6] Moved undeliveredMessageInterceptor to local variable to avoid threading issues. --- Source/MQTTnet/Server/MqttClientSessionsManager.cs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index a0f0bf2e..d27a366f 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -238,12 +238,14 @@ namespace MQTTnet.Server if (deliveryCount == 0) { - if (_options.UndeliveredMessageInterceptor == null) + var undeliveredMessageInterceptor = _options.UndeliveredMessageInterceptor; + + if (undeliveredMessageInterceptor == null) { return; } - await _options.UndeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(sender?.ClientId, sender?.Session?.Items, applicationMessage)); + await undeliveredMessageInterceptor.InterceptApplicationMessagePublishAsync(new MqttApplicationMessageInterceptorContext(sender?.ClientId, sender?.Session?.Items, applicationMessage)); } } catch (OperationCanceledException) From 6adf7096ec3a9da627bf5768e87b4c816d8638dc Mon Sep 17 00:00:00 2001 From: SeppPenner Date: Fri, 12 Jun 2020 20:25:51 +0200 Subject: [PATCH 6/6] Updated client sessions manager to handle the delivery count for the handler correctly. --- Source/MQTTnet/Server/MqttClientSession.cs | 6 ++++-- Source/MQTTnet/Server/MqttClientSessionsManager.cs | 7 +++++-- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index c7d2e50d..c6c3e0ef 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -46,17 +46,19 @@ namespace MQTTnet.Server /// public IDictionary Items { get; } - public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage) + public bool EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage) { var checkSubscriptionsResult = SubscriptionsManager.CheckSubscriptions(applicationMessage.Topic, applicationMessage.QualityOfServiceLevel); if (!checkSubscriptionsResult.IsSubscribed) { - return; + return true; } _logger.Verbose("Queued application message with topic '{0}' (ClientId: {1}).", applicationMessage.Topic, ClientId); ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage); + + return false; } public async Task SubscribeAsync(ICollection topicFilters) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index d27a366f..8a34f7b1 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -228,12 +228,15 @@ namespace MQTTnet.Server foreach (var clientSession in _sessions.Values) { - clientSession.EnqueueApplicationMessage( + var isSubscribed = clientSession.EnqueueApplicationMessage( applicationMessage, sender?.ClientId, false); - deliveryCount++; + if (isSubscribed) + { + deliveryCount++; + } } if (deliveryCount == 0)