diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index b366be73..3d25a3da 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -18,6 +18,7 @@ * [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). * [Core] Fixed a null reference exception in the MqttTcpChannel with WriteAsync and ReadAsync. +* [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/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..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 5ac84638..48f69809 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -220,12 +220,31 @@ namespace MQTTnet.Server await _retainedMessagesManager.HandleMessageAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false); } + var deliveryCount = 0; + foreach (var clientSession in _sessions.Values) { - clientSession.EnqueueApplicationMessage( + var isSubscribed = clientSession.EnqueueApplicationMessage( applicationMessage, sender?.ClientId, false); + + if (isSubscribed) + { + deliveryCount++; + } + } + + if (deliveryCount == 0) + { + var undeliveredMessageInterceptor = _options.UndeliveredMessageInterceptor; + + if (undeliveredMessageInterceptor == null) + { + return; + } + + await 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"); + + } + } } }