Add interceptor for all packets.

Source/MQTTnet.AspnetCore/Extensions/ServiceCollectionExtensions.cs

@@ -67,7 +67,7 @@ namespace MQTTnet.AspNetCore.Extensions
             services.AddSingleton<IMqttNetLogger>(logger);
             services.AddSingleton<MqttHostedServer>();
             services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
-            services.AddSingleton<IMqttServer>(s => s.GetService<MqttHostedServer>());
+            services.AddSingleton<MqttServer>(s => s.GetService<MqttHostedServer>());
 
             return services;
         }

Source/MQTTnet.AspnetCore/MqttHostedServer.cs

@@ -3,82 +3,27 @@ using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
-using MQTTnet.Client.Publishing;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Server;
 
 namespace MQTTnet.AspNetCore
 {
-    public sealed class MqttHostedServer : IMqttServer, IHostedService
+    public sealed class MqttHostedServer : MqttServer, IHostedService
     {
-        readonly MqttServer _mqttServer;
-
         public MqttHostedServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) 
+            : base(options, adapters, logger)
         {
-            _mqttServer = new MqttServer(options, adapters, logger);
         }
 
         public Task StartAsync(CancellationToken cancellationToken)
         {
-            _ = _mqttServer.StartAsync();
+            _ =  StartAsync();
             return Task.CompletedTask;
         }
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
-            return _mqttServer.StopAsync();
+            return StopAsync();
-        }
-        
-        public Task<MqttClientPublishResult> PublishAsync(string senderClientId, MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
-        {
-            return _mqttServer.PublishAsync(senderClientId, applicationMessage, cancellationToken);
-        }
-
-        public void Dispose()
-        {
-            _mqttServer.Dispose();
-        }
-
-        public bool IsStarted => _mqttServer.IsStarted;
-        
-        public Task<IList<IMqttClientStatus>> GetClientStatusAsync()
-        {
-            return _mqttServer.GetClientStatusAsync();
-        }
-
-        public Task<IList<IMqttSessionStatus>> GetSessionStatusAsync()
-        {
-            return _mqttServer.GetSessionStatusAsync();
-        }
-
-        public Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync()
-        {
-            return _mqttServer.GetRetainedApplicationMessagesAsync();
-        }
-
-        public Task ClearRetainedApplicationMessagesAsync()
-        {
-            return _mqttServer.ClearRetainedApplicationMessagesAsync();
-        }
-
-        public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
-        {
-            return _mqttServer.SubscribeAsync(clientId, topicFilters);
-        }
-
-        public Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters)
-        {
-            return _mqttServer.UnsubscribeAsync(clientId, topicFilters);
-        }
-
-        public Task StartAsync()
-        {
-            return _mqttServer.StartAsync();
-        }
-
-        public Task StopAsync()
-        {
-            return _mqttServer.StopAsync();
         }
     }
 }

Source/MQTTnet/MQTTnet.csproj.DotSettings

@@ -1,3 +1,5 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cevents/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Coptions/@EntryIndexedValue">True</s:Boolean>
-	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstatus/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstatus/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstorage/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs

@@ -0,0 +1,30 @@
+using System;
+using System.Threading;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public sealed class InterceptingPacketEventArgs : EventArgs
+    {
+        /// <summary>
+        /// Gets the client ID which has sent the packet or will receive the packet.
+        /// </summary>
+        public string ClientId { get; internal set; }
+        
+        /// <summary>
+        /// Gets the endpoint of the sending or receiving client.
+        /// </summary>
+        public string Endpoint { get; internal set; }
+        
+        /// <summary>
+        /// Gets or sets the MQTT packet which was received or will be sent.
+        /// </summary>
+        public MqttBasePacket Packet { get; set; }
+        
+        /// <summary>
+        /// Gets the cancellation token from the connection managing thread.
+        /// Use this in further event processing.
+        /// </summary>
+        public CancellationToken CancellationToken { get; internal set; }
+    }
+}

Source/MQTTnet/Server/IMqttServer.cs

@@ -1,23 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Threading.Tasks;
-
-namespace MQTTnet.Server
-{
-    public interface IMqttServer : IDisposable
-    {
-        bool IsStarted { get; }
-        
-        Task<IList<IMqttClientStatus>> GetClientStatusAsync();
-        Task<IList<IMqttSessionStatus>> GetSessionStatusAsync();
-
-        Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync();
-        Task ClearRetainedApplicationMessagesAsync();
-
-        Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters);
-        Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters);
-
-        Task StartAsync();
-        Task StopAsync();
-    }
-}

Source/MQTTnet/Server/Internal/MqttClientConnection.cs

@@ -27,6 +27,7 @@ namespace MQTTnet.Server.Internal
         readonly MqttApplicationMessageInterceptorInvoker _applicationMessageInterceptorInvoker;
         
         readonly MqttServerOptions _serverOptions;
+        readonly MqttServerEventContainer _eventContainer;
         readonly MqttConnectPacket _connectPacket;
 
         CancellationTokenSource _cancellationToken;
@@ -41,6 +42,7 @@ namespace MQTTnet.Server.Internal
             IMqttNetLogger logger)
         {
             _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
+            _eventContainer = eventContainer;
             _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
 
             ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
@@ -155,6 +157,18 @@ namespace MQTTnet.Server.Internal
                 while (!cancellationToken.IsCancellationRequested)
                 {
                     var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
+
+                    var interceptingPacketEventArgs = new InterceptingPacketEventArgs
+                    {
+                        ClientId = Id,
+                        Endpoint = Endpoint,
+                        Packet = packet,
+                        CancellationToken = cancellationToken
+                    };
+                    
+                    await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false);
+                    packet = interceptingPacketEventArgs.Packet;
+                    
                     if (packet == null)
                     {
                         // The client has closed the connection gracefully.

Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs

@@ -31,6 +31,10 @@ namespace MQTTnet.Server.Internal
         
         public AsyncEvent<MqttApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<MqttApplicationMessageNotConsumedEventArgs>();
         
+        public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
+        
+        public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
+        
         public AsyncEvent<EventArgs> StartedEvent { get; } = new AsyncEvent<EventArgs>();
         
         public AsyncEvent<EventArgs> StoppedEvent { get; } = new AsyncEvent<EventArgs>();

Source/MQTTnet/Server/MqttServer.cs

@@ -14,7 +14,7 @@ using MQTTnet.Server.Internal;
 
 namespace MQTTnet.Server
 {
-    public class MqttServer : Disposable, IMqttServer
+    public class MqttServer : Disposable
     {
         readonly MqttServerEventContainer _eventContainer;
         
@@ -71,6 +71,18 @@ namespace MQTTnet.Server
             remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value);
         }
         
+        public event Func<InterceptingPacketEventArgs, Task> InterceptInboundPacketAsync
+        {
+            add => _eventContainer.InterceptingInboundPacketEvent.AddHandler(value);
+            remove => _eventContainer.InterceptingInboundPacketEvent.RemoveHandler(value);
+        }
+        
+        public event Func<InterceptingPacketEventArgs, Task> InterceptingOutboundPacketAsync
+        {
+            add => _eventContainer.InterceptingOutboundPacketEvent.AddHandler(value);
+            remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value);
+        }
+        
         public event Func<InterceptingMqttClientSubscriptionEventArgs, Task> InterceptingClientSubscriptionAsync
         {
             add => _eventContainer.InterceptingClientSubscriptionEvent.AddHandler(value);

Source/MQTTnet/Server/MqttServerExtensions.cs

@@ -8,7 +8,7 @@ namespace MQTTnet.Server
 {
     public static class MqttServerExtensions
     {
-        public static Task SubscribeAsync(this IMqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
+        public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
         {
             if (server == null) throw new ArgumentNullException(nameof(server));
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));
@@ -17,7 +17,7 @@ namespace MQTTnet.Server
             return server.SubscribeAsync(clientId, topicFilters);
         }
 
-        public static Task SubscribeAsync(this IMqttServer server, string clientId, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
+        public static Task SubscribeAsync(this MqttServer server, string clientId, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
         {
             if (server == null) throw new ArgumentNullException(nameof(server));
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));
@@ -26,7 +26,7 @@ namespace MQTTnet.Server
             return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
         }
 
-        public static Task SubscribeAsync(this IMqttServer server, string clientId, string topic)
+        public static Task SubscribeAsync(this MqttServer server, string clientId, string topic)
         {
             if (server == null) throw new ArgumentNullException(nameof(server));
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));
@@ -35,7 +35,7 @@ namespace MQTTnet.Server
             return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build());
         }
 
-        public static Task UnsubscribeAsync(this IMqttServer server, string clientId, params string[] topicFilters)
+        public static Task UnsubscribeAsync(this MqttServer server, string clientId, params string[] topicFilters)
         {
             if (server == null) throw new ArgumentNullException(nameof(server));
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));

Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs

@@ -1,5 +1,4 @@
-using MQTTnet.Diagnostics;
+using System.Collections.Generic;
-using System.Collections.Generic;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 

Tests/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs

@@ -12,7 +12,7 @@ namespace MQTTnet.Benchmarks
     [MemoryDiagnoser]
     public sealed class MqttTcpChannelBenchmark
     {
-        IMqttServer _mqttServer;
+        MqttServer _mqttServer;
         IMqttChannel _serverChannel;
         IMqttChannel _clientChannel;