Add new APIs.

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)
         {

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;

Source/MQTTnet/Server/MqttServerOptionsBuilder.cs

@@ -218,6 +218,12 @@ namespace MQTTnet.Server
             _options.SubscriptionInterceptor = value;
             return this;
         }
+        
+        public MqttServerOptionsBuilder WithSubscriptionInterceptor(Action<MqttSubscriptionInterceptorContext> 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<MqttSubscriptionInterceptorContext> value)
+        public MqttServerOptionsBuilder WithUnsubscriptionInterceptor(Action<MqttUnsubscriptionInterceptorContext> value)
         {
-            _options.SubscriptionInterceptor = new MqttServerSubscriptionInterceptorDelegate(value);
+            _options.UnsubscriptionInterceptor = new MqttServerUnsubscriptionInterceptorDelegate(value);
             return this;
         }
 

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<MqttUnsubscriptionInterceptorContext, Task> _callback;
+
+        public MqttServerUnsubscriptionInterceptorDelegate(Action<MqttUnsubscriptionInterceptorContext> callback)
+        {
+            if (callback == null) throw new ArgumentNullException(nameof(callback));
+
+            _callback = context =>
+            {
+                callback(context);
+                return PlatformAbstractionLayer.CompletedTask;
+            };
+        }
+
+        public MqttServerUnsubscriptionInterceptorDelegate(Func<MqttUnsubscriptionInterceptorContext, Task> callback)
+        {
+            _callback = callback ?? throw new ArgumentNullException(nameof(callback));
+        }
+
+        public Task InterceptUnsubscriptionAsync(MqttUnsubscriptionInterceptorContext context)
+        {
+            return _callback(context);
+        }
+    }
+}

Source/MQTTnet/Server/Status/IMqttSessionStatus.cs

@@ -34,6 +34,13 @@ namespace MQTTnet.Server.Status
         /// <returns>A task.</returns>
         Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage);
         
+        /// <summary>
+        /// Delivers a new application message.
+        /// </summary>
+        /// <param name="applicationMessage">The application message.</param>
+        /// <returns>A task.</returns>
+        Task DeliverApplicationMessageAsync(MqttApplicationMessage applicationMessage);
+        
         /// <summary>
         /// Clears the entire queue with pending application messages.
         /// </summary>

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();

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();