From 369b262e07f19c1ad1e82d64be22a0a01a28dc7b Mon Sep 17 00:00:00 2001 From: Christian Date: Wed, 27 Oct 2021 22:10:42 +0200 Subject: [PATCH] Restore undelivered event. --- .../Internal/MqttClientSessionsManager.cs | 23 ++++---------- .../Internal/MqttServerEventContainer.cs | 3 ++ ...tApplicationMessageNotConsumedEventArgs.cs | 17 +++++++++++ Source/MQTTnet/Server/MqttServer.cs | 6 ++++ Tests/MQTTnet.Core.Tests/Server/General.cs | 30 +++++++++---------- 5 files changed, 47 insertions(+), 32 deletions(-) create mode 100644 Source/MQTTnet/Server/MqttApplicationMessageNotConsumedEventArgs.cs diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 0c6333f8..4cb27609 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -407,14 +407,7 @@ namespace MQTTnet.Server.Internal } _logger.Verbose("Client '{0}': Queued application message with topic '{1}'.", clientSession.Id, applicationMessage.Topic); - - // var queuedApplicationMessage = new MqttQueuedApplicationMessage - // { - // ApplicationMessage = applicationMessage, - // SubscriptionQualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel, - // SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers - // }; - + var newPublishPacket = _packetFactories.Publish.Create(applicationMessage); newPublishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel; newPublishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers); @@ -441,15 +434,11 @@ namespace MQTTnet.Server.Internal if (deliveryCount == 0) { - // TODO: Addd event for not delivered publish packets. - // var undeliveredMessageInterceptor = _options.UndeliveredMessageInterceptor; - // if (undeliveredMessageInterceptor == null) - // { - // return; - // } - // - // // The delegate signature is the same as for regular message interceptor. So the call is fine and just uses a different interceptor. - // await InterceptApplicationMessageAsync(undeliveredMessageInterceptor, sender, applicationMessage).ConfigureAwait(false); + await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(() => new MqttApplicationMessageNotConsumedEventArgs + { + ApplicationMessage = applicationMessage, + SenderClientId = senderClientId + }).ConfigureAwait(false); } } catch (Exception exception) diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs index 5d050d0f..3d494079 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs @@ -28,6 +28,9 @@ namespace MQTTnet.Server.Internal public AsyncEvent ClientUnsubscribedTopicEvent { get; } = new AsyncEvent(); public AsyncEvent PreparingClientSessionEvent { get; } = new AsyncEvent(); + + public AsyncEvent ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent(); + public AsyncEvent StartedEvent { get; } = new AsyncEvent(); public AsyncEvent StoppedEvent { get; } = new AsyncEvent(); diff --git a/Source/MQTTnet/Server/MqttApplicationMessageNotConsumedEventArgs.cs b/Source/MQTTnet/Server/MqttApplicationMessageNotConsumedEventArgs.cs new file mode 100644 index 00000000..bbc09675 --- /dev/null +++ b/Source/MQTTnet/Server/MqttApplicationMessageNotConsumedEventArgs.cs @@ -0,0 +1,17 @@ +using System; + +namespace MQTTnet.Server +{ + public sealed class MqttApplicationMessageNotConsumedEventArgs : EventArgs + { + /// + /// Gets the application message which was not consumed by any client. + /// + public MqttApplicationMessage ApplicationMessage { get; internal set; } + + /// + /// Gets the ID of the client which has sent the affected application message. + /// + public string SenderClientId { get; internal set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 227d9e04..4ff21c88 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -107,6 +107,12 @@ namespace MQTTnet.Server remove => _eventContainer.PreparingClientSessionEvent.RemoveHandler(value); } + public event Func ApplicationMessageNotConsumedAsync + { + add => _eventContainer.ApplicationMessageNotConsumedEvent.AddHandler(value); + remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value); + } + public bool IsStarted => _cancellationTokenSource != null; public Task> GetClientStatusAsync() diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index 8a37aa95..88b300e6 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -1027,22 +1027,22 @@ namespace MQTTnet.Tests.Server { var undeliverd = string.Empty; - throw new NotImplementedException(); + var server = await testEnvironment.StartServer(); + server.ApplicationMessageNotConsumedAsync += e => + { + undeliverd = e.ApplicationMessage.Topic; + return Task.CompletedTask; + }; - // var options = new MqttServerOptionsBuilder().WithUndeliveredMessageInterceptor( - // context => { undeliverd = context.ApplicationMessage.Topic; }); - // - // await testEnvironment.StartServer(options); - // - // var client = await testEnvironment.ConnectClient(); - // - // await client.SubscribeAsync("b"); - // - // await client.PublishAsync("a", null, MqttQualityOfServiceLevel.ExactlyOnce); - // - // await Task.Delay(500); - // - // Assert.AreEqual("a", undeliverd); + var client = await testEnvironment.ConnectClient(); + + await client.SubscribeAsync("b"); + + await client.PublishAsync("a", null, MqttQualityOfServiceLevel.ExactlyOnce); + + await Task.Delay(500); + + Assert.AreEqual("a", undeliverd); } }