From 212d88f16752915845086085d4d2c256574e1757 Mon Sep 17 00:00:00 2001 From: Christian Date: Thu, 21 Oct 2021 19:15:57 +0200 Subject: [PATCH] Add new APIs. --- Source/MQTTnet/Internal/MqttPacketBus.cs | 7 ---- .../Internal/MqttServerKeepAliveMonitor.cs | 1 - .../Server/MqttServerOptionsBuilder.cs | 10 ++++-- ...ServerUnsubscriptionInterceptorDelegate.cs | 32 +++++++++++++++++++ .../Server/Status/IMqttSessionStatus.cs | 7 ++++ .../Server/Status/MqttSessionStatus.cs | 7 ++++ .../Internal/MqttPacketBus_Tests.cs | 12 +++++-- 7 files changed, 63 insertions(+), 13 deletions(-) create mode 100644 Source/MQTTnet/Server/MqttServerUnsubscriptionInterceptorDelegate.cs diff --git a/Source/MQTTnet/Internal/MqttPacketBus.cs b/Source/MQTTnet/Internal/MqttPacketBus.cs index 43313d0d..683a9d9a 100644 --- a/Source/MQTTnet/Internal/MqttPacketBus.cs +++ b/Source/MQTTnet/Internal/MqttPacketBus.cs @@ -48,13 +48,6 @@ namespace MQTTnet.Internal return _partitions[(int) partition].Select(i => i.Packet).ToList(); } } - - public MqttPacketBusItem Enqueue(MqttBasePacket packet, MqttPacketBusPartition partition) - { - var busItem = new MqttPacketBusItem(packet); - Enqueue(busItem, partition); - return busItem; - } public void Enqueue(MqttPacketBusItem item, MqttPacketBusPartition partition) { diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index 72f50056..66211c2d 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -2,7 +2,6 @@ using System.Threading; using System.Threading.Tasks; using MQTTnet.Client.Disconnecting; -using MQTTnet.Diagnostics; using MQTTnet.Diagnostics.Logger; using MQTTnet.Implementations; using MQTTnet.Internal; diff --git a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs index c56a6b81..0f2d6aea 100644 --- a/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs +++ b/Source/MQTTnet/Server/MqttServerOptionsBuilder.cs @@ -218,6 +218,12 @@ namespace MQTTnet.Server _options.SubscriptionInterceptor = value; return this; } + + public MqttServerOptionsBuilder WithSubscriptionInterceptor(Action value) + { + _options.SubscriptionInterceptor = new MqttServerSubscriptionInterceptorDelegate(value); + return this; + } public MqttServerOptionsBuilder WithUnsubscriptionInterceptor(IMqttServerUnsubscriptionInterceptor value) { @@ -225,9 +231,9 @@ namespace MQTTnet.Server return this; } - public MqttServerOptionsBuilder WithSubscriptionInterceptor(Action value) + public MqttServerOptionsBuilder WithUnsubscriptionInterceptor(Action value) { - _options.SubscriptionInterceptor = new MqttServerSubscriptionInterceptorDelegate(value); + _options.UnsubscriptionInterceptor = new MqttServerUnsubscriptionInterceptorDelegate(value); return this; } diff --git a/Source/MQTTnet/Server/MqttServerUnsubscriptionInterceptorDelegate.cs b/Source/MQTTnet/Server/MqttServerUnsubscriptionInterceptorDelegate.cs new file mode 100644 index 00000000..d70f1e14 --- /dev/null +++ b/Source/MQTTnet/Server/MqttServerUnsubscriptionInterceptorDelegate.cs @@ -0,0 +1,32 @@ +using System; +using System.Threading.Tasks; +using MQTTnet.Implementations; + +namespace MQTTnet.Server +{ + public sealed class MqttServerUnsubscriptionInterceptorDelegate : IMqttServerUnsubscriptionInterceptor + { + readonly Func _callback; + + public MqttServerUnsubscriptionInterceptorDelegate(Action callback) + { + if (callback == null) throw new ArgumentNullException(nameof(callback)); + + _callback = context => + { + callback(context); + return PlatformAbstractionLayer.CompletedTask; + }; + } + + public MqttServerUnsubscriptionInterceptorDelegate(Func callback) + { + _callback = callback ?? throw new ArgumentNullException(nameof(callback)); + } + + public Task InterceptUnsubscriptionAsync(MqttUnsubscriptionInterceptorContext context) + { + return _callback(context); + } + } +} diff --git a/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs b/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs index 6d66b11a..f7a5b4a6 100644 --- a/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/IMqttSessionStatus.cs @@ -34,6 +34,13 @@ namespace MQTTnet.Server.Status /// A task. Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage); + /// + /// Delivers a new application message. + /// + /// The application message. + /// A task. + Task DeliverApplicationMessageAsync(MqttApplicationMessage applicationMessage); + /// /// Clears the entire queue with pending application messages. /// diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index 9fdeee05..0ee53a76 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -45,6 +45,13 @@ namespace MQTTnet.Server.Status return PlatformAbstractionLayer.CompletedTask; } + public Task DeliverApplicationMessageAsync(MqttApplicationMessage applicationMessage) + { + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); + + throw new NotImplementedException(); + } + public Task ClearApplicationMessagesQueueAsync() { _session.ApplicationMessagesQueue.Clear(); diff --git a/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs b/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs index 465ad1db..aa177f91 100644 --- a/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs @@ -83,14 +83,20 @@ namespace MQTTnet.Tests.Internal var bus = new MqttPacketBus(); var delivered = false; + + var item1 = new MqttPacketBusItem(new MqttPublishPacket()); + var item2 = new MqttPacketBusItem(new MqttPublishPacket()); - bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data); - bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data); - bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data).Delivered += (_, __) => + var item3 = new MqttPacketBusItem(new MqttPublishPacket()); + item3.Delivered += (_, __) => { delivered = true; }; + bus.Enqueue(item1, MqttPacketBusPartition.Data); + bus.Enqueue(item2, MqttPacketBusPartition.Data); + bus.Enqueue(item3, MqttPacketBusPartition.Data); + Assert.IsFalse(delivered); bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered();