Add more handling options to interceptor contexts.

Build/MQTTnet.nuspec

@@ -14,6 +14,7 @@
     <releaseNotes>
 * [Core] Added all builders to the MQTT factory.
 * [Core] Removed global logger and refactored logging (BREAKING CHANGE!).
+* [Core] Improved memory management when working with large payloads.
 * [Client] Renamed _MqttClientConnectResult_ to _MqttClientConnectResult_ (BREAKING CHANGE!).
 * [Client] The OS will now choose the best TLS version to use. It is no longer fixed to 1.3 etc. (thanks to @patagonaa, #1271).
 * [Client] Exposed user properties and reason string in subscribe result.
@@ -35,6 +36,7 @@
 * [Server] Added support for returning individual subscription errors (#80 thanks to @jimch)
 * [Server] Improved topic filter comparisons (support for $).
 * [Server] Fixed wrong "IsSessionPresent" result (thanks to @logicaloud, #1268).
+* [Server] Added more MQTTv5 response information to all interceptors (BREAKING CHANGE!).
 * [MQTTnet.Server] Moved server project to a dedicated GitHub repository.
 Git commit: $gitCommit
     </releaseNotes>

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -173,7 +173,7 @@ namespace MQTTnet.AspNetCore
             using (await _writerLock.WaitAsync(cancellationToken).ConfigureAwait(false))
             {
                 var buffer = formatter.Encode(packet);
-                var msg = buffer.AsMemory();
+                var msg = buffer.ToArray().AsMemory();
                 var output = _output;
                 var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
                 if (result.IsCompleted)

Source/MQTTnet.AspnetCore/MqttHostedServer.cs

@@ -4,30 +4,137 @@ using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
+using MQTTnet.Client.Publishing;
+using MQTTnet.Client.Receiving;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Server;
+using MQTTnet.Server.Status;
 
 namespace MQTTnet.AspNetCore
 {
-    public sealed class MqttHostedServer : MqttServer, IHostedService
+    public sealed class MqttHostedServer : IMqttServer, IHostedService
     {
+        readonly MqttServer _mqttServer;
         readonly IMqttServerOptions _options;
 
-        public MqttHostedServer(IMqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) 
-            : base(adapters, logger)
+        public MqttHostedServer(IMqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
+
+            _mqttServer = new MqttServer(adapters, logger);
         }
 
         public Task StartAsync(CancellationToken cancellationToken)
         {
-            _ = StartAsync(_options);
+            _ = _mqttServer.StartAsync(_options);
             return Task.CompletedTask;
         }
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
-            return StopAsync();
+            return _mqttServer.StopAsync();
+        }
+
+        public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
+        {
+            get => _mqttServer.ApplicationMessageReceivedHandler;
+            set => _mqttServer.ApplicationMessageReceivedHandler = value;
+        }
+
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => _mqttServer.ApplicationMessageReceivedAsync += value;
+            remove => _mqttServer.ApplicationMessageReceivedAsync -= value;
+        }
+        
+        public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
+        {
+            return _mqttServer.PublishAsync(applicationMessage, cancellationToken);
+        }
+
+        public void Dispose()
+        {
+            _mqttServer.Dispose();
+        }
+
+        public bool IsStarted => _mqttServer.IsStarted;
+
+        public IMqttServerStartedHandler StartedHandler
+        {
+            get => _mqttServer.StartedHandler;
+            set => _mqttServer.StartedHandler = value;
+        }
+
+        public IMqttServerStoppedHandler StoppedHandler
+        {
+            get => _mqttServer.StoppedHandler;
+            set => _mqttServer.StoppedHandler = value;
+        }
+
+        public IMqttServerClientConnectedHandler ClientConnectedHandler
+        {
+            get => _mqttServer.ClientConnectedHandler;
+            set => _mqttServer.ClientConnectedHandler = value;
+        }
+
+        public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler
+        {
+            get => _mqttServer.ClientDisconnectedHandler;
+            set => _mqttServer.ClientDisconnectedHandler = value;
+        }
+
+        public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler
+        {
+            get => _mqttServer.ClientSubscribedTopicHandler;
+            set => _mqttServer.ClientSubscribedTopicHandler = value;
+        }
+
+        public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler
+        {
+            get => _mqttServer.ClientUnsubscribedTopicHandler;
+            set => _mqttServer.ClientUnsubscribedTopicHandler = value;
+        }
+
+        public IMqttServerOptions Options => _mqttServer.Options;
+
+        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(IMqttServerOptions options)
+        {
+            return _mqttServer.StartAsync(options);
+        }
+
+        public Task StopAsync()
+        {
+            return _mqttServer.StopAsync();
         }
     }
-}
+}

Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs

@@ -17,10 +17,10 @@ using MQTTnet.Diagnostics.Logger;
 
 namespace MQTTnet.Extensions.ManagedClient
 {
-    public class ManagedMqttClient : Disposable, IManagedMqttClient
+    public sealed class ManagedMqttClient : Disposable, IManagedMqttClient
     {
         readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
-
+        
         /// <summary>
         /// The subscriptions are managed in 2 separate buckets:
         /// <see cref="_subscriptions"/> and <see cref="_unsubscriptions"/> are processed during normal operation
@@ -29,7 +29,6 @@ namespace MQTTnet.Extensions.ManagedClient
         ///  at reconnect and are solely owned by <see cref="MaintainConnectionAsync"/>.
         /// </summary>
         readonly Dictionary<string, MqttQualityOfServiceLevel> _reconnectSubscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
-
         readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
         readonly HashSet<string> _unsubscriptions = new HashSet<string>();
         readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);
@@ -80,6 +79,12 @@ namespace MQTTnet.Extensions.ManagedClient
             set => InternalClient.ApplicationMessageReceivedHandler = value;
         }
 
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => InternalClient.ApplicationMessageReceivedAsync += value;
+            remove => InternalClient.ApplicationMessageReceivedAsync -= value;
+        }
+
         public IApplicationMessageProcessedHandler ApplicationMessageProcessedHandler { get; set; }
 
         public IApplicationMessageSkippedHandler ApplicationMessageSkippedHandler { get; set; }

Source/MQTTnet/Adapter/MqttChannelAdapter.cs

@@ -116,11 +116,11 @@ namespace MQTTnet.Adapter
                     var packetData = PacketFormatterAdapter.Encode(packet);
                     _packetInspectorHandler.BeginSendPacket(packetData);
 
-                    await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false);
+                    await packetData.WriteToAsync(_channel, cancellationToken).ConfigureAwait(false);
+                    
+                    Interlocked.Add(ref _bytesReceived, packetData.Length);
 
-                    Interlocked.Add(ref _bytesReceived, packetData.Count);
-
-                    _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet);
+                    _logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Length, packet);
                 }
                 catch (Exception exception)
                 {

Source/MQTTnet/Adapter/MqttPacketInspectorHandler.cs

@@ -4,6 +4,7 @@ using System.Linq;
 using MQTTnet.Diagnostics;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Diagnostics.PacketInspection;
+using MQTTnet.Formatter;
 
 namespace MQTTnet.Adapter
 {
@@ -44,7 +45,7 @@ namespace MQTTnet.Adapter
             InspectPacket(buffer, MqttPacketFlowDirection.Inbound);
         }
 
-        public void BeginSendPacket(ArraySegment<byte> buffer)
+        public void BeginSendPacket(MqttPacketBuffer buffer)
         {
             if (_packetInspector == null)
             {
@@ -56,7 +57,7 @@ namespace MQTTnet.Adapter
             // intended for debugging etc. so that this is OK.
             var bufferCopy = buffer.ToArray();
 
-            InspectPacket(bufferCopy, MqttPacketFlowDirection.Outbound);
+            InspectPacket(bufferCopy.ToArray(), MqttPacketFlowDirection.Outbound);
         }
 
         public void FillReceiveBuffer(byte[] buffer)

Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs

@@ -12,7 +12,9 @@ namespace MQTTnet.Client.Disconnecting
             ConnectResult = connectResult;
             Reason = reason;
 
+#pragma warning disable 618
             ReasonCode = reason;
+#pragma warning restore 618
         }
 
         public bool ClientWasConnected { get; }

Source/MQTTnet/Client/IMqttClient.cs

@@ -22,17 +22,27 @@ namespace MQTTnet.Client
         /// </summary>
         IMqttClientConnectedHandler ConnectedHandler { get; set; }
 
+        /// <summary>
+        /// Fired when the client is connected.
+        /// </summary>
+        event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;
+        
         /// <summary>
         /// Gets or sets the disconnected handler that is fired after the client has disconnected from the server.
         /// Hint: Initialize handlers before you connect the client to avoid issues.
         /// </summary>
         IMqttClientDisconnectedHandler DisconnectedHandler { get; set; }
 
-        Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken);
+        /// <summary>
+        /// Fired when the client is disconnected.
+        /// </summary>
+        event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync;
+        
+        Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken = default);
 
-        Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken);
+        Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default);
 
-        Task PingAsync(CancellationToken cancellationToken);
+        Task PingAsync(CancellationToken cancellationToken = default);
 
         /// <summary>
         /// Sends extended authentication data.
@@ -41,10 +51,10 @@ namespace MQTTnet.Client
         /// <param name="data">The extended data.</param>
         /// <param name="cancellationToken">A cancellation token to stop the task.</param>
         /// <returns>A <see cref="Task"/> representing any asynchronous operation.</returns>
-        Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken);
+        Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default);
 
-        Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken);
+        Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default);
 
-        Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken);
+        Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default);
     }
 }

Source/MQTTnet/Client/MqttClient.cs

@@ -21,13 +21,17 @@ using MQTTnet.Implementations;
 
 namespace MQTTnet.Client
 {
-    public class MqttClient : Disposable, IMqttClient
+    public sealed class MqttClient : Disposable, IMqttClient
     {
         readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
         readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
         readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
         readonly object _disconnectLock = new object();
 
+        readonly AsyncEvent<MqttClientConnectedEventArgs> _connectedEvent = new AsyncEvent<MqttClientConnectedEventArgs>();
+        readonly AsyncEvent<MqttClientDisconnectedEventArgs> _disconnectedEvent = new AsyncEvent<MqttClientDisconnectedEventArgs>();
+        readonly AsyncEvent<MqttApplicationMessageReceivedEventArgs> _applicationMessageReceivedEvent = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
+
         readonly IMqttClientAdapterFactory _adapterFactory;
         readonly MqttNetSourceLogger _logger;
 
@@ -56,10 +60,28 @@ namespace MQTTnet.Client
 
         public IMqttClientConnectedHandler ConnectedHandler { get; set; }
 
+        public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
+        {
+            add => _connectedEvent.AddHandler(value);
+            remove => _connectedEvent.RemoveHandler(value);
+        }
+        
         public IMqttClientDisconnectedHandler DisconnectedHandler { get; set; }
 
+        public event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync
+        {
+            add => _disconnectedEvent.AddHandler(value);
+            remove => _disconnectedEvent.RemoveHandler(value);
+        }
+        
         public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
 
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => _applicationMessageReceivedEvent.AddHandler(value);
+            remove => _applicationMessageReceivedEvent.RemoveHandler(value);
+        }
+
         public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
 
         public IMqttClientOptions Options { get; private set; }
@@ -118,12 +140,15 @@ namespace MQTTnet.Client
 
                 _logger.Info("Connected.");
 
+                var eventArgs = new MqttClientConnectedEventArgs(connectResult);
                 var connectedHandler = ConnectedHandler;
                 if (connectedHandler != null)
                 {
                     await connectedHandler.HandleConnectedAsync(new MqttClientConnectedEventArgs(connectResult)).ConfigureAwait(false);
                 }
 
+                await _connectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+
                 return connectResult;
             }
             catch (Exception exception)
@@ -391,13 +416,17 @@ namespace MQTTnet.Client
 
                 _logger.Info("Disconnected.");
 
+                var eventArgs = new MqttClientDisconnectedEventArgs(clientWasConnected, exception, connectResult, _disconnectReason);
+                
                 var disconnectedHandler = DisconnectedHandler;
                 if (disconnectedHandler != null)
                 {
                     // This handler must be executed in a new thread because otherwise a dead lock may happen
                     // when trying to reconnect in that handler etc.
-                    Task.Run(() => disconnectedHandler.HandleDisconnectedAsync(new MqttClientDisconnectedEventArgs(clientWasConnected, exception, connectResult, _disconnectReason))).RunInBackground(_logger);
+                    Task.Run(() => disconnectedHandler.HandleDisconnectedAsync(eventArgs)).RunInBackground(_logger);
                 }
+
+                Task.Run(() => _disconnectedEvent.InvokeAsync(eventArgs)).RunInBackground(_logger);
             }
         }
 
@@ -791,6 +820,8 @@ namespace MQTTnet.Client
                 await handler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false);
             }
 
+            await _applicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
+
             return eventArgs;
         }
 

Source/MQTTnet/Client/MqttClientExtensions.cs

@@ -149,8 +149,19 @@ namespace MQTTnet.Client
 
             return client.DisconnectAsync(options, CancellationToken.None);
         }
+        
+        public static Task DisconnectAsync(this IMqttClient client, MqttClientDisconnectReason reasonCode, string reasonString = null)
+        {
+            if (client == null) throw new ArgumentNullException(nameof(client));
 
-        public static Task DisconnectAsync(this IMqttClient client, CancellationToken cancellationToken)
+            return client.DisconnectAsync(new MqttClientDisconnectOptions
+            {
+                ReasonCode = reasonCode,
+                ReasonString = reasonString
+            }, CancellationToken.None);
+        }
+
+        public static Task DisconnectAsync(this IMqttClient client, CancellationToken cancellationToken = default)
         {
             if (client == null) throw new ArgumentNullException(nameof(client));
 
@@ -194,43 +205,14 @@ namespace MQTTnet.Client
 
             return client.UnsubscribeAsync(options);
         }
-
-        public static Task<MqttClientConnectResult> ConnectAsync(this IMqttClient client, IMqttClientOptions options)
-        {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-
-            return client.ConnectAsync(options, CancellationToken.None);
-        }
-
+        
         public static Task SendExtendedAuthenticationExchangeDataAsync(this IMqttClient client, MqttExtendedAuthenticationExchangeData data)
         {
             if (client == null) throw new ArgumentNullException(nameof(client));
 
             return client.SendExtendedAuthenticationExchangeDataAsync(data, CancellationToken.None);
         }
-
-        public static Task<MqttClientSubscribeResult> SubscribeAsync(this IMqttClient client, MqttClientSubscribeOptions options)
-        {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-
-            return client.SubscribeAsync(options, CancellationToken.None);
-        }
-
-        public static Task<MqttClientUnsubscribeResult> UnsubscribeAsync(this IMqttClient client, MqttClientUnsubscribeOptions options)
-        {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-
-            return client.UnsubscribeAsync(options, CancellationToken.None);
-        }
-
-        public static Task<MqttClientPublishResult> PublishAsync(this IMqttClient client, MqttApplicationMessage applicationMessage)
-        {
-            if (client == null) throw new ArgumentNullException(nameof(client));
-            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
-
-            return client.PublishAsync(applicationMessage, CancellationToken.None);
-        }
-
+        
         public static async Task PublishAsync(this IMqttClient client, IEnumerable<MqttApplicationMessage> applicationMessages)
         {
             if (client == null) throw new ArgumentNullException(nameof(client));

Source/MQTTnet/Client/Subscribing/MqttClientSubscribeResultFactory.cs

@@ -12,13 +12,6 @@ namespace MQTTnet.Client.Subscribing
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
             if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket));
             
-            // MQTTv3.1.1 handling.
-            if (subAckPacket.ReturnCodes.Any() && subAckPacket.ReturnCodes.Count != subscribePacket.TopicFilters.Count)
-            {
-                throw new MqttProtocolViolationException(
-                    "The return codes are not matching the topic filters [MQTT-3.9.3-1].");
-            }
-            
             // MQTTv5.0.0 handling.
             if (subAckPacket.ReasonCodes.Any() && subAckPacket.ReasonCodes.Count != subscribePacket.TopicFilters.Count)
             {
@@ -43,19 +36,8 @@ namespace MQTTnet.Client.Subscribing
 
         static MqttClientSubscribeResultItem CreateSubscribeResultItem(int index, MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket)
         {
-            MqttClientSubscribeResultCode resultCode;
-            
-            if (subAckPacket.ReturnCodes.Any())
-            {
-                // MQTTv3.1.1 handling.
-                resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReturnCodes[index];
-            }
-            else
-            {
-                // MQTTv5.0.0 handling.
-                resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[index];
-            }
-            
+            var resultCode = (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[index];
+
             return new MqttClientSubscribeResultItem
             {
                 TopicFilter = subscribePacket.TopicFilters[index],

Source/MQTTnet/Formatter/IMqttPacketFormatter.cs

@@ -1,12 +1,11 @@
-using System;
-using MQTTnet.Adapter;
+using MQTTnet.Adapter;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Formatter
 {
     public interface IMqttPacketFormatter
     {
-        ArraySegment<byte> Encode(MqttBasePacket mqttPacket);
+        MqttPacketBuffer Encode(MqttBasePacket mqttPacket);
 
         MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket);
 

Source/MQTTnet/Formatter/MqttPacketBuffer.cs

@@ -0,0 +1,51 @@
+using System;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Channel;
+
+namespace MQTTnet.Formatter
+{
+    public sealed class MqttPacketBuffer
+    {
+        readonly ArraySegment<byte>[] _segments;
+
+        public MqttPacketBuffer(params ArraySegment<byte>[] segments)
+        {
+            _segments = segments ?? throw new ArgumentNullException(nameof(segments));
+
+            Length = _segments.Sum(s => s.Count);
+        }
+
+        public int Length { get; }
+        
+        public ArraySegment<byte> ToArray()
+        {
+            if (_segments.Length == 1)
+            {
+                return _segments[0];
+            }
+            
+            using (var buffer = new MemoryStream(Length))
+            {
+                foreach (var segment in _segments)
+                {
+                    buffer.Write(segment.Array, segment.Offset, segment.Count);
+                }
+
+                return new ArraySegment<byte>(buffer.ToArray());    
+            }
+        }
+
+        public async Task WriteToAsync(IMqttChannel channel, CancellationToken cancellationToken)
+        {
+            if (channel == null) throw new ArgumentNullException(nameof(channel));
+            
+            foreach (var segment in _segments)
+            {
+                await channel.WriteAsync(segment.Array, segment.Offset, segment.Count, cancellationToken).ConfigureAwait(false);
+            }
+        }
+    }
+}

Source/MQTTnet/Formatter/MqttPacketFormatterAdapter.cs

@@ -32,7 +32,7 @@ namespace MQTTnet.Formatter
         
         public IMqttPacketWriter Writer { get; }
 
-        public ArraySegment<byte> Encode(MqttBasePacket packet)
+        public MqttPacketBuffer Encode(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 

Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs

@@ -1,25 +1,48 @@
 using System;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
 
 namespace MQTTnet.Formatter
 {
     public sealed class MqttPubAckPacketFactory
     {
+        public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse)
+        {
+            if (applicationMessageResponse == null) throw new ArgumentNullException(nameof(applicationMessageResponse));
+
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
+            var pubAckPacket = new MqttPubAckPacket
+            {
+                PacketIdentifier = publishPacket.PacketIdentifier,
+                ReasonCode = (MqttPubAckReasonCode) (int) applicationMessageResponse.ReasonCode,
+                Properties =
+                {
+                    ReasonString = applicationMessageResponse.ReasonString
+                }
+            };
+
+            pubAckPacket.Properties.UserProperties.AddRange(applicationMessageResponse.UserProperties);
+            
+            return pubAckPacket;
+        }
+        
         public MqttPubAckPacket Create(MqttApplicationMessageReceivedEventArgs applicationMessageReceivedEventArgs)
         {
             if (applicationMessageReceivedEventArgs == null) throw new ArgumentNullException(nameof(applicationMessageReceivedEventArgs));
 
             var pubAckPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode);
             pubAckPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties);
-            
+            pubAckPacket.Properties.ReasonString = applicationMessageReceivedEventArgs.ResponseReasonString;
+
             return pubAckPacket;
         }
 
-        public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode)
+        static MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode)
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
-            
+
             var pubAckPacket = new MqttPubAckPacket
             {
                 PacketIdentifier = publishPacket.PacketIdentifier,

Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs

@@ -1,9 +1,7 @@
-
-
-
-using System;
+using System;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
+using MQTTnet.Server;
 
 namespace MQTTnet.Formatter
 {
@@ -15,20 +13,38 @@ namespace MQTTnet.Formatter
 
             var pubRecPacket = Create(applicationMessageReceivedEventArgs.PublishPacket, applicationMessageReceivedEventArgs.ReasonCode);
             pubRecPacket.Properties.UserProperties.AddRange(applicationMessageReceivedEventArgs.ResponseUserProperties);
-            
+
             return pubRecPacket;
         }
 
-        public MqttPubRecPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode)
+        public MqttBasePacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse)
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
-            
+
+            var pubRecPacket = new MqttPubRecPacket
+            {
+                PacketIdentifier = publishPacket.PacketIdentifier,
+                ReasonCode = (MqttPubRecReasonCode) (int) applicationMessageResponse.ReasonCode,
+                Properties =
+                {
+                    ReasonString = applicationMessageResponse.ReasonString
+                }
+            };
+
+            pubRecPacket.Properties.UserProperties.AddRange(applicationMessageResponse.UserProperties);
+            return pubRecPacket;
+        }
+
+        MqttPubRecPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode applicationMessageReceivedReasonCode)
+        {
+            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+
             var pubRecPacket = new MqttPubRecPacket
             {
                 PacketIdentifier = publishPacket.PacketIdentifier,
                 ReasonCode = (MqttPubRecReasonCode) (int) applicationMessageReceivedReasonCode
             };
-            
+
             return pubRecPacket;
         }
     }

Source/MQTTnet/Formatter/MqttPublishPacketFactory.cs

@@ -29,7 +29,7 @@ namespace MQTTnet.Formatter
                     TopicAlias = applicationMessage.TopicAlias
                 }
             };
-
+            
             if (applicationMessage.SubscriptionIdentifiers != null)
             {
                 packet.Properties.SubscriptionIdentifiers.AddRange(applicationMessage.SubscriptionIdentifiers);    
@@ -73,5 +73,18 @@ namespace MQTTnet.Formatter
 
             return packet;
         }
+
+        public MqttPublishPacket Clone(MqttPublishPacket publishPacket)
+        {
+            return new MqttPublishPacket
+            {
+                Topic = publishPacket.Topic,
+                Payload = publishPacket.Payload,
+                Retain = publishPacket.Retain,
+                QualityOfServiceLevel = publishPacket.QualityOfServiceLevel,
+                Dup = publishPacket.Dup,
+                PacketIdentifier = publishPacket.PacketIdentifier
+            };
+        }
     }
 }

Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs

@@ -6,21 +6,17 @@ namespace MQTTnet.Formatter
 {
     public sealed class MqttSubAckPacketFactory
     {
-        public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult)
+        public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, MqttSubscribeResult mqttSubscribeResult)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
-            if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult));
+            if (mqttSubscribeResult == null) throw new ArgumentNullException(nameof(mqttSubscribeResult));
 
             var subAckPacket = new MqttSubAckPacket
             {
                 PacketIdentifier = subscribePacket.PacketIdentifier
             };
 
-            // MQTTv3.1.1
-            subAckPacket.ReturnCodes.AddRange(subscribeResult.ReturnCodes);
-            
-            // MQTTv5.0.0
-            subAckPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes);
+            subAckPacket.ReasonCodes.AddRange(mqttSubscribeResult.ReasonCodes);
             
             return subAckPacket;
         }

Source/MQTTnet/Formatter/MqttUnsubAckPacketFactory.cs

@@ -6,10 +6,10 @@ namespace MQTTnet.Formatter
 {
     public sealed class MqttUnsubAckPacketFactory
     {
-        public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult)
+        public MqttUnsubAckPacket Create(MqttUnsubscribePacket unsubscribePacket, MqttUnsubscribeResult mqttUnsubscribeResult)
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
-            if (unsubscribeResult == null) throw new ArgumentNullException(nameof(unsubscribeResult));
+            if (mqttUnsubscribeResult == null) throw new ArgumentNullException(nameof(mqttUnsubscribeResult));
 
             var unsubAckPacket = new MqttUnsubAckPacket
             {
@@ -17,7 +17,7 @@ namespace MQTTnet.Formatter
             };
 
             // MQTTv5.0.0 only.
-            unsubAckPacket.ReasonCodes.AddRange(unsubscribeResult.ReasonCodes);
+            unsubAckPacket.ReasonCodes.AddRange(mqttUnsubscribeResult.ReasonCodes);
             
             return unsubAckPacket;
         }

Source/MQTTnet/Formatter/V3/MqttV310PacketFormatter.cs

@@ -21,7 +21,7 @@ namespace MQTTnet.Formatter.V3
             _packetWriter = packetWriter;
         }
 
-        public ArraySegment<byte> Encode(MqttBasePacket packet)
+        public MqttPacketBuffer Encode(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
@@ -32,6 +32,12 @@ namespace MQTTnet.Formatter.V3
             var fixedHeader = EncodePacket(packet, _packetWriter);
             var remainingLength = (uint) (_packetWriter.Length - 5);
 
+            var publishPacket = packet as MqttPublishPacket;
+            if (publishPacket?.Payload != null)
+            {
+                remainingLength += (uint) publishPacket.Payload.Length;
+            }
+
             var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
 
             var headerSize = FixedHeaderSize + remainingLengthSize;
@@ -44,7 +50,15 @@ namespace MQTTnet.Formatter.V3
 
             var buffer = _packetWriter.GetBuffer();
 
-            return new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
+            var firstSegment = new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
+            
+            if (publishPacket?.Payload != null)
+            {
+                var payloadSegment = new ArraySegment<byte>(publishPacket.Payload, 0, publishPacket.Payload.Length);
+                return new MqttPacketBuffer(firstSegment, payloadSegment);   
+            }
+
+            return new MqttPacketBuffer(firstSegment);
         }
 
         public MqttBasePacket Decode(ReceivedMqttPacket receivedMqttPacket)
@@ -300,7 +314,7 @@ namespace MQTTnet.Formatter.V3
 
             while (!body.EndOfStream)
             {
-                packet.ReturnCodes.Add((MqttSubscribeReturnCode) body.ReadByte());
+                packet.ReasonCodes.Add((MqttSubscribeReasonCode) body.ReadByte());
             }
 
             return packet;
@@ -442,10 +456,8 @@ namespace MQTTnet.Formatter.V3
                 }
             }
 
-            if (packet.Payload?.Length > 0)
-            {
-                packetWriter.Write(packet.Payload, 0, packet.Payload.Length);
-            }
+            // The payload is the past part of the packet. But it is not added here in order to keep
+            // memory allocation low.
 
             byte fixedHeader = 0;
 
@@ -547,11 +559,26 @@ namespace MQTTnet.Formatter.V3
 
             packetWriter.Write(packet.PacketIdentifier);
 
-            if (packet.ReturnCodes?.Any() == true)
+            if (packet.ReasonCodes.Any())
             {
-                foreach (var packetSubscribeReturnCode in packet.ReturnCodes)
+                foreach (var packetSubscribeReturnCode in packet.ReasonCodes)
                 {
-                    packetWriter.Write((byte) packetSubscribeReturnCode);
+                    if (packetSubscribeReturnCode == MqttSubscribeReasonCode.GrantedQoS0)
+                    {
+                        packetWriter.Write((byte) MqttSubscribeReturnCode.SuccessMaximumQoS0);    
+                    }
+                    else if (packetSubscribeReturnCode == MqttSubscribeReasonCode.GrantedQoS1)
+                    {
+                        packetWriter.Write((byte) MqttSubscribeReturnCode.SuccessMaximumQoS1);    
+                    }
+                    else if (packetSubscribeReturnCode == MqttSubscribeReasonCode.GrantedQoS2)
+                    {
+                        packetWriter.Write((byte) MqttSubscribeReturnCode.SuccessMaximumQoS2);    
+                    }
+                    else
+                    {
+                        packetWriter.Write((byte)MqttSubscribeReturnCode.Failure);
+                    }
                 }
             }
 

Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs

@@ -15,7 +15,7 @@ namespace MQTTnet.Formatter.V5
             _packetWriter = packetWriter ?? throw new ArgumentNullException(nameof(packetWriter));
         }
 
-        public ArraySegment<byte> Encode(MqttBasePacket packet)
+        public MqttPacketBuffer Encode(MqttBasePacket packet)
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
@@ -24,8 +24,14 @@ namespace MQTTnet.Formatter.V5
             _packetWriter.Seek(5);
 
             var fixedHeader = EncodePacket(packet, _packetWriter);
-            var remainingLength = (uint)(_packetWriter.Length - 5);
+            var remainingLength = (uint)_packetWriter.Length - 5;
 
+            var publishPacket = packet as MqttPublishPacket;
+            if (publishPacket?.Payload != null)
+            {
+                remainingLength += (uint)publishPacket.Payload.Length;
+            }
+            
             var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
             
             var headerSize = 1 + remainingLengthSize;
@@ -37,7 +43,16 @@ namespace MQTTnet.Formatter.V5
             _packetWriter.WriteVariableLengthInteger(remainingLength);
 
             var buffer = _packetWriter.GetBuffer();
-            return new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
+
+            var firstSegment = new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
+            
+            if (publishPacket?.Payload != null)
+            {
+                var payloadSegment = new ArraySegment<byte>(publishPacket.Payload, 0, publishPacket.Payload.Length);
+                return new MqttPacketBuffer(firstSegment, payloadSegment);   
+            }
+            
+            return new MqttPacketBuffer(firstSegment);
         }
 
         public void FreeBuffer()
@@ -256,10 +271,8 @@ namespace MQTTnet.Formatter.V5
 
             propertiesWriter.WriteTo(packetWriter);
 
-            if (packet.Payload?.Length > 0)
-            {
-                packetWriter.Write(packet.Payload, 0, packet.Payload.Length);
-            }
+            // The payload is the past part of the packet. But it is not added here in order to keep
+            // memory allocation low.
 
             byte fixedHeader = 0;
 

Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs

@@ -14,7 +14,7 @@ namespace MQTTnet.Formatter.V5
             _encoder = new MqttV500PacketEncoder(writer);
         }
         
-        public ArraySegment<byte> Encode(MqttBasePacket mqttPacket)
+        public MqttPacketBuffer Encode(MqttBasePacket mqttPacket)
         {
             if (mqttPacket == null) throw new ArgumentNullException(nameof(mqttPacket));
 

Source/MQTTnet/IApplicationMessagePublisher.cs

@@ -6,6 +6,6 @@ namespace MQTTnet
 {
     public interface IApplicationMessagePublisher
     {
-        Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken);
+        Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default);
     }
 }

Source/MQTTnet/IApplicationMessageReceiver.cs

@@ -1,4 +1,6 @@
-using MQTTnet.Client.Receiving;
+using System;
+using System.Threading.Tasks;
+using MQTTnet.Client.Receiving;
 
 namespace MQTTnet
 {
@@ -9,5 +11,10 @@ namespace MQTTnet
         /// Hint: Initialize handlers before you connect the client to avoid issues.
         /// </summary>
         IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
+
+        /// <summary>
+        /// Fired when an application message was received.
+        /// </summary>
+        event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
     }
 }

Source/MQTTnet/Internal/AsyncEvent.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Internal
+{
+    public sealed class AsyncEvent<TEventArgs>
+    {
+        readonly List<Func<TEventArgs, Task>> _handlers = new List<Func<TEventArgs, Task>>();
+
+        public void AddHandler(Func<TEventArgs, Task> handler)
+        {
+            if (handler == null) throw new ArgumentNullException(nameof(handler));
+            
+            _handlers.Add(handler);
+        }
+
+        public void RemoveHandler(Func<TEventArgs, Task> handler)
+        {
+            if (handler == null) throw new ArgumentNullException(nameof(handler));
+            
+            _handlers.Remove(handler);
+        }
+
+        public async Task InvokeAsync(TEventArgs eventArgs)
+        {
+            foreach (var handler in _handlers)
+            {
+                await handler.Invoke(eventArgs);
+            }
+        }
+    }
+}

Source/MQTTnet/MqttApplicationMessageBuilder.cs

@@ -40,9 +40,7 @@ namespace MQTTnet
         /// The broker stores the last retained message and the corresponding QoS for that topic.
         /// </summary>
         bool _retain;
-
-        bool _dup;
-
+        
         /// <summary>
         /// The content type.
         /// The content type must be a UTF-8 encoded string. The content type value identifies the kind of UTF-8 encoded payload.
@@ -198,13 +196,7 @@ namespace MQTTnet
             _retain = value;
             return this;
         }
-
-        public MqttApplicationMessageBuilder WithDupFlag(bool value = true)
-        {
-            _dup = value;
-            return this;
-        }
-
+        
         public MqttApplicationMessageBuilder WithAtLeastOnceQoS()
         {
             _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
@@ -340,19 +332,13 @@ namespace MQTTnet
             {
                 throw new MqttProtocolViolationException("A Topic Alias of 0 is not permitted. A sender MUST NOT send a PUBLISH packet containing a Topic Alias which has the value 0 [MQTT-3.3.2-8].");
             }
-
-            if (_qualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce && _dup)
-            {
-                throw new MqttProtocolViolationException("The DUP flag MUST be set to 0 for all QoS 0 messages [MQTT-3.3.1-2].");
-            }
-
+            
             var applicationMessage = new MqttApplicationMessage
             {
                 Topic = _topic,
                 Payload = _payload,
                 QualityOfServiceLevel = _qualityOfServiceLevel,
                 Retain = _retain,
-                Dup = _dup,
                 ContentType = _contentType,
                 ResponseTopic = _responseTopic,
                 CorrelationData = _correlationData,

Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs

@@ -24,7 +24,7 @@ namespace MQTTnet
             _acknowledgeHandler = acknowledgeHandler;
         }
 
-        internal MqttPublishPacket PublishPacket { get; }
+        internal MqttPublishPacket PublishPacket { get; set; }
         
         /// <summary>
         /// Gets the client identifier.
@@ -36,13 +36,21 @@ namespace MQTTnet
 
         public bool ProcessingFailed { get; set; }
 
+        /// <summary>
+        /// Gets or sets the reason code which will be sent to the server.
+        /// </summary>
         public MqttApplicationMessageReceivedReasonCode ReasonCode { get; set; } = MqttApplicationMessageReceivedReasonCode.Success;
 
         /// <summary>
-        /// Gets or sets the user properties which being sent to the server in the ACK packet etc.
+        /// Gets or sets the user properties which will be sent to the server in the ACK packet etc.
         /// </summary>
         public List<MqttUserProperty> ResponseUserProperties { get; } = new List<MqttUserProperty>();
 
+        /// <summary>
+        /// Gets or sets the reason string which will be sent to the server in the ACK packet.
+        /// </summary>
+        public string ResponseReasonString { get; set; }
+        
         /// <summary>
         /// Gets or sets whether this message was handled.
         /// This value can be used in user code for custom control flow.

Source/MQTTnet/Packets/MqttSubAckPacket.cs

@@ -7,14 +7,10 @@ namespace MQTTnet.Packets
     public sealed class MqttSubAckPacket : MqttBasePacket, IMqttPacketWithIdentifier
     {
         public ushort PacketIdentifier { get; set; }
-
-        /// <summary>
-        /// Only available in MQTT v3.1.1.
-        /// </summary>
-        public List<MqttSubscribeReturnCode> ReturnCodes { get; } = new List<MqttSubscribeReturnCode>();
         
         /// <summary>
-        /// Added in MQTTv5.
+        /// Reason Code is used in MQTTv5.0.0 and backward compatible to v.3.1.1.
+        /// Return Code is used in MQTTv3.1.1
         /// </summary>
         public List<MqttSubscribeReasonCode> ReasonCodes { get; } = new List<MqttSubscribeReasonCode>();
 
@@ -25,10 +21,9 @@ namespace MQTTnet.Packets
 
         public override string ToString()
         {
-            var returnCodesText = string.Join(",", ReturnCodes.Select(f => f.ToString()));
             var reasonCodesText = string.Join(",", ReasonCodes.Select(f => f.ToString()));
 
-            return string.Concat("SubAck: [PacketIdentifier=", PacketIdentifier, "] [ReturnCodes=", returnCodesText, "] [ReasonCode=", reasonCodesText, "]");
+            return string.Concat("SubAck: [PacketIdentifier=", PacketIdentifier, "] [ReasonCode=", reasonCodesText, "]");
         }
     }
 }

Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs

@@ -0,0 +1,63 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Server.Internal
+{
+    public sealed class MqttApplicationMessageInterceptorInvoker
+    {
+        readonly IMqttServerOptions _serverOptions;
+        readonly string _clientId;
+        readonly IDictionary<object, object> _sessionItems;
+        
+        public MqttApplicationMessageInterceptorInvoker(IMqttServerOptions serverOptions, string clientId, IDictionary<object, object> sessionItems)
+        {
+            _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
+            _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
+            _sessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems));
+        }
+
+        public MqttApplicationMessage ApplicationMessage { get; private set; }
+        
+        public MqttApplicationMessageResponse Response { get; private set; } = new MqttApplicationMessageResponse();
+            
+        public bool ProcessPublish { get; private set; } = true;
+
+        public bool CloseConnection { get; private set; }
+            
+        public async Task Invoke(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
+        {
+            // Reset to defaults.
+            Response = new MqttApplicationMessageResponse();
+            ProcessPublish = true;
+            CloseConnection = false;
+            ApplicationMessage = applicationMessage;
+                
+            // Intercept.
+            var interceptor = _serverOptions.ApplicationMessageInterceptor;
+            if (interceptor == null)
+            {
+                return;
+            }
+            
+            var interceptorContext = new MqttApplicationMessageInterceptorContext
+            {
+                ClientId = _clientId,
+                ApplicationMessage = applicationMessage,
+                SessionItems = _sessionItems,
+                ProcessPublish = true,
+                CloseConnection = false,
+                CancellationToken = cancellationToken
+            };
+
+            await interceptor.InterceptApplicationMessagePublishAsync(interceptorContext).ConfigureAwait(false);
+
+            // Expose results.
+            ProcessPublish = interceptorContext.ProcessPublish && interceptorContext.ApplicationMessage != null;
+            CloseConnection = interceptorContext.CloseConnection;
+            Response = interceptorContext.Response;
+            ApplicationMessage = interceptorContext.ApplicationMessage;
+        }
+    }
+}

Source/MQTTnet/Server/Internal/MqttClientConnection.cs

@@ -7,7 +7,6 @@ using MQTTnet.Client.Disconnecting;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
-using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.PacketDispatcher;
 using MQTTnet.Packets;
@@ -20,16 +19,18 @@ namespace MQTTnet.Server.Internal
         readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
         readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
         readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
+        readonly MqttApplicationMessageFactory _applicationMessageFactory = new MqttApplicationMessageFactory();
         
         readonly MqttClientSessionsManager _sessionsManager;
-
         readonly MqttNetSourceLogger _logger;
+        
+        readonly MqttApplicationMessageInterceptorInvoker _applicationMessageInterceptorInvoker;
+        
         readonly IMqttServerOptions _serverOptions;
-
         readonly MqttConnectPacket _connectPacket;
 
         CancellationTokenSource _cancellationToken;
-
+        
         public MqttClientConnection(
             MqttConnectPacket connectPacket,
             IMqttChannelAdapter channelAdapter,
@@ -47,6 +48,8 @@ namespace MQTTnet.Server.Internal
             Session = session ?? throw new ArgumentNullException(nameof(session));
             _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
 
+            _applicationMessageInterceptorInvoker = new MqttApplicationMessageInterceptorInvoker(_serverOptions, Id, Session.Items);
+            
             if (logger == null) throw new ArgumentNullException(nameof(logger));
             _logger = logger.WithSource(nameof(MqttClientConnection));
         }
@@ -109,10 +112,7 @@ namespace MQTTnet.Server.Internal
                 try
                 {
                     Task.Run(() => SendPacketsLoop(cancellationToken.Token), cancellationToken.Token).RunInBackground(_logger);
-
-                    // TODO: Check!.
-                    //Session.IsCleanSession = false;
-
+                    
                     IsRunning = true;
 
                     await ReceivePackagesLoop(cancellationToken.Token).ConfigureAwait(false);
@@ -130,8 +130,9 @@ namespace MQTTnet.Server.Internal
 
             if (!IsTakenOver && !IsCleanDisconnect && Session.LatestConnectPacket.WillFlag && !Session.WillMessageSent)
             {
-                var publishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket);
-                _sessionsManager.DispatchPublishPacket(publishPacket, this);
+                var willPublishPacket = _packetFactories.Publish.Create(Session.LatestConnectPacket);
+                var willApplicationMessage = _applicationMessageFactory.Create(willPublishPacket);
+                _ = _sessionsManager.DispatchPublishPacket(willApplicationMessage, this);
                 Session.WillMessageSent = true;
             }
 
@@ -163,7 +164,7 @@ namespace MQTTnet.Server.Internal
 
                     if (packet is MqttPublishPacket publishPacket)
                     {
-                        HandleIncomingPublishPacket(publishPacket);
+                        await HandleIncomingPublishPacket(publishPacket, cancellationToken).ConfigureAwait(false);
                     }
                     else if (packet is MqttPubAckPacket pubAckPacket)
                     {
@@ -329,7 +330,7 @@ namespace MQTTnet.Server.Internal
                         Session.EnqueuePacket(new MqttPacketBusItem(publishPacket));
                     }
                 }
-                
+
                 StopInternal();
             }
         }
@@ -366,7 +367,7 @@ namespace MQTTnet.Server.Internal
         {
             _cancellationToken?.Cancel();
         }
-        
+
         void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket)
         {
             var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success);
@@ -413,12 +414,27 @@ namespace MQTTnet.Server.Internal
                 StopInternal();
             }
         }
-
-        void HandleIncomingPublishPacket(MqttPublishPacket publishPacket)
+       
+        async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             HandleTopicAlias(publishPacket);
 
-            _sessionsManager.DispatchPublishPacket(publishPacket, this);
+            var applicationMessage = new MqttApplicationMessageFactory().Create(publishPacket);
+
+            await _applicationMessageInterceptorInvoker.Invoke(applicationMessage, cancellationToken).ConfigureAwait(false);
+
+            applicationMessage = _applicationMessageInterceptorInvoker.ApplicationMessage;
+            
+            if (_applicationMessageInterceptorInvoker.CloseConnection)
+            {
+                await StopAsync(MqttClientDisconnectReason.UnspecifiedError);
+                return;
+            }
+
+            if (_applicationMessageInterceptorInvoker.ProcessPublish)
+            {
+                await _sessionsManager.DispatchPublishPacket(applicationMessage, this).ConfigureAwait(false);    
+            }
             
             switch (publishPacket.QualityOfServiceLevel)
             {
@@ -429,13 +445,13 @@ namespace MQTTnet.Server.Internal
                 }
                 case MqttQualityOfServiceLevel.AtLeastOnce:
                 {
-                    var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
+                    var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, _applicationMessageInterceptorInvoker.Response);
                     Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket));
                     break;
                 }
                 case MqttQualityOfServiceLevel.ExactlyOnce:
                 {
-                    var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, MqttApplicationMessageReceivedReasonCode.Success);
+                    var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, _applicationMessageInterceptorInvoker.Response);
                     Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket));
                     break;
                 }
@@ -497,37 +513,39 @@ namespace MQTTnet.Server.Internal
             }
         }
 
-        // TODO: Fix!
-        async Task<MqttPublishPacket> InvokeClientMessageQueueInterceptor(MqttPublishPacket publishPacket, MqttQueuedApplicationMessage queuedApplicationMessage)
-        {
-            if (_serverOptions.ClientMessageQueueInterceptor == null)
-            {
-                return publishPacket;
-            }
+        // // TODO: Fix!
+        // async Task<MqttPublishPacket> InvokeClientMessageQueueInterceptor(MqttPublishPacket publishPacket, MqttQueuedApplicationMessage queuedApplicationMessage)
+        // {
+        //     if (_serverOptions.ClientMessageQueueInterceptor == null)
+        //     {
+        //         return publishPacket;
+        //     }
+        //
+        //     var context = new MqttClientMessageQueueInterceptorContext
+        //     {
+        //         SenderClientId = queuedApplicationMessage.SenderClientId,
+        //         ReceiverClientId = Id,
+        //         ApplicationMessage = queuedApplicationMessage.ApplicationMessage,
+        //         SubscriptionQualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel
+        //     };
+        //
+        //     if (_serverOptions.ClientMessageQueueInterceptor != null)
+        //     {
+        //         await _serverOptions.ClientMessageQueueInterceptor.InterceptClientMessageQueueEnqueueAsync(context).ConfigureAwait(false);
+        //     }
+        //
+        //     if (!context.AcceptEnqueue || context.ApplicationMessage == null)
+        //     {
+        //         return null;
+        //     }
+        //
+        //     publishPacket.Topic = context.ApplicationMessage.Topic;
+        //     publishPacket.Payload = context.ApplicationMessage.Payload;
+        //     publishPacket.QualityOfServiceLevel = context.SubscriptionQualityOfServiceLevel;
+        //
+        //     return publishPacket;
+        // }
+        //
 
-            var context = new MqttClientMessageQueueInterceptorContext
-            {
-                SenderClientId = queuedApplicationMessage.SenderClientId,
-                ReceiverClientId = Id,
-                ApplicationMessage = queuedApplicationMessage.ApplicationMessage,
-                SubscriptionQualityOfServiceLevel = queuedApplicationMessage.SubscriptionQualityOfServiceLevel
-            };
-
-            if (_serverOptions.ClientMessageQueueInterceptor != null)
-            {
-                await _serverOptions.ClientMessageQueueInterceptor.InterceptClientMessageQueueEnqueueAsync(context).ConfigureAwait(false);
-            }
-
-            if (!context.AcceptEnqueue || context.ApplicationMessage == null)
-            {
-                return null;
-            }
-
-            publishPacket.Topic = context.ApplicationMessage.Topic;
-            publishPacket.Payload = context.ApplicationMessage.Payload;
-            publishPacket.QualityOfServiceLevel = context.SubscriptionQualityOfServiceLevel;
-
-            return publishPacket;
-        }
     }
 }

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -16,9 +16,9 @@ using MQTTnet.Server.Status;
 
 namespace MQTTnet.Server.Internal
 {
-    public sealed class MqttClientSessionsManager : IDisposable
+    public sealed class MqttClientSessionsManager
     {
-        readonly BlockingCollection<MqttPendingApplicationMessage> _messageQueue = new BlockingCollection<MqttPendingApplicationMessage>();
+        //readonly BlockingCollection<MqttPendingApplicationMessage> _messageQueue = new BlockingCollection<MqttPendingApplicationMessage>();
         readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
         
         readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
@@ -51,10 +51,10 @@ namespace MQTTnet.Server.Internal
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
-        public void Start(CancellationToken cancellationToken)
-        {
-            Task.Run(() => TryProcessQueuedApplicationMessagesAsync(cancellationToken), cancellationToken).RunInBackground(_logger);
-        }
+        // public void Start(CancellationToken cancellationToken)
+        // {
+        //     Task.Run(() => TryProcessQueuedApplicationMessagesAsync(cancellationToken), cancellationToken).RunInBackground(_logger);
+        // }
 
         async Task<MqttConnectPacket> ReceiveConnectPacket(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
         {
@@ -225,13 +225,13 @@ namespace MQTTnet.Server.Internal
             return Task.FromResult((IList<IMqttSessionStatus>)result);
         }
 
-        public void DispatchPublishPacket(MqttPublishPacket publishPacket, MqttClientConnection sender)
-        {
-            if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
-
-            _messageQueue.Add(new MqttPendingApplicationMessage(publishPacket, sender));
-        }
-
+        // public void DispatchPublishPacket(MqttPublishPacket publishPacket, MqttClientConnection sender)
+        // {
+        //     if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
+        //
+        //     _messageQueue.Add(new MqttPendingApplicationMessage(publishPacket, sender));
+        // }
+        
         public async Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
         {
             if (clientId == null) throw new ArgumentNullException(nameof(clientId));
@@ -304,80 +304,82 @@ namespace MQTTnet.Server.Internal
             _logger.Verbose("Session for client '{0}' deleted.", clientId);
         }
 
-        public void Dispose()
-        {
-            _messageQueue?.Dispose();
-        }
+        // public void Dispose()
+        // {
+        //     _messageQueue?.Dispose();
+        // }
 
-        async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)
-        {
-            // Make sure all queued messages are processed before server stops.
-            while (!cancellationToken.IsCancellationRequested || _messageQueue.Any())
-            {
-                try
-                {
-                    await TryProcessNextQueuedApplicationMessage().ConfigureAwait(false);
-                }
-                catch (OperationCanceledException)
-                {
-                }
-                catch (Exception exception)
-                {
-                    _logger.Error(exception, "Unhandled exception while processing queued application messages.");
-                }
-            }
-        }
+        // async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)
+        // {
+        //     // Make sure all queued messages are processed before server stops.
+        //     while (!cancellationToken.IsCancellationRequested || _messageQueue.Any())
+        //     {
+        //         try
+        //         {
+        //             await TryProcessNextQueuedApplicationMessage().ConfigureAwait(false);
+        //         }
+        //         catch (OperationCanceledException)
+        //         {
+        //         }
+        //         catch (Exception exception)
+        //         {
+        //             _logger.Error(exception, "Unhandled exception while processing queued application messages.");
+        //         }
+        //     }
+        // }
 
-        async Task TryProcessNextQueuedApplicationMessage()
+        public async Task DispatchPublishPacket(MqttApplicationMessage applicationMessage, MqttClientConnection sender)
         {
             try
             {
-                MqttPendingApplicationMessage pendingApplicationMessage;
-                try
-                {
-                    pendingApplicationMessage = _messageQueue.Take();
-                }
-                catch (ArgumentNullException)
-                {
-                    return;
-                }
-                // catch (ThreadAbortException)
+                var senderClientId = sender?.Id ?? _options.ClientId;
+       
+                // MqttPendingApplicationMessage pendingApplicationMessage;
+                // try
+                // {
+                //     pendingApplicationMessage = _messageQueue.Take();
+                // }
+                // catch (ArgumentNullException)
+                // {
+                //     return;
+                // }
+                // // catch (ThreadAbortException)
+                // // {
+                // //     return;
+                // // }
+                // catch (ObjectDisposedException)
                 // {
                 //     return;
                 // }
-                catch (ObjectDisposedException)
-                {
-                    return;
-                }
 
-                var clientConnection = pendingApplicationMessage.Sender;
-                var senderClientId = clientConnection?.Id ?? _options.ClientId;
+                //var clientConnection = pendingApplicationMessage.Sender;
+                //var senderClientId = clientConnection?.Id ?? _options.ClientId;
                 
-                // TODO: Only use PUBLISH packets internally!
-                var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket);
-
-                var interceptor = _options.ApplicationMessageInterceptor;
-                if (interceptor != null)
-                {
-                    var interceptorContext = await InterceptApplicationMessageAsync(interceptor, clientConnection, applicationMessage).ConfigureAwait(false);
-                    if (interceptorContext != null)
-                    {
-                        if (interceptorContext.CloseConnection)
-                        {
-                            if (clientConnection != null)
-                            {
-                                await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
-                            }
-                        }
-
-                        if (interceptorContext.ApplicationMessage == null || !interceptorContext.ProcessPublish)
-                        {
-                            return;
-                        }
-
-                        applicationMessage = interceptorContext.ApplicationMessage;
-                    }
-                }
+                // // TODO: Only use PUBLISH packets internally!
+                // var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket);
+                //
+                // var interceptor = _options.ApplicationMessageInterceptor;
+                // if (interceptor != null)
+                // {
+                //     var interceptorContext = await InterceptApplicationMessageAsync(interceptor, clientConnection, applicationMessage).ConfigureAwait(false);
+                //     if (interceptorContext != null)
+                //     {
+                //         if (interceptorContext.CloseConnection)
+                //         {
+                //             if (clientConnection != null)
+                //             {
+                //                 await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
+                //             }
+                //         }
+                //
+                //         if (interceptorContext.ApplicationMessage == null || !interceptorContext.ProcessPublish)
+                //         {
+                //             return;
+                //         }
+                //
+                //         applicationMessage = interceptorContext.ApplicationMessage;
+                //     }
+                // }
 
                 await _eventDispatcher.SafeNotifyApplicationMessageReceivedAsync(senderClientId, applicationMessage).ConfigureAwait(false);
 
@@ -414,28 +416,27 @@ namespace MQTTnet.Server.Internal
                     //     SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers
                     // };
                    
-                    var publishPacket = _packetFactories.Publish.Create(applicationMessage);
-                    
-                    publishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
-                    publishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers);
+                    var newPublishPacket = _packetFactories.Publish.Create(applicationMessage);
+                    newPublishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
+                    newPublishPacket.Properties.SubscriptionIdentifiers.AddRange(checkSubscriptionsResult.SubscriptionIdentifiers);
 
-                    if (publishPacket.QualityOfServiceLevel > 0)
+                    if (newPublishPacket.QualityOfServiceLevel > 0)
                     {
-                        publishPacket.PacketIdentifier = clientSession.PacketIdentifierProvider.GetNextPacketIdentifier();
+                        newPublishPacket.PacketIdentifier = clientSession.PacketIdentifierProvider.GetNextPacketIdentifier();
                     }
                     
                     if (checkSubscriptionsResult.RetainAsPublished)
                     {
                         // Transfer the original retain state from the publisher.
                         // This is a MQTTv5 feature.
-                        publishPacket.Retain = applicationMessage.Retain;
+                        newPublishPacket.Retain = applicationMessage.Retain;
                     }
                     else
                     {
-                        publishPacket.Retain = false;
+                        newPublishPacket.Retain = false;
                     }
                     
-                    clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket));
+                    clientSession.EnqueuePacket(new MqttPacketBusItem(newPublishPacket));
                     deliveryCount++;
                 }
 
@@ -448,7 +449,7 @@ namespace MQTTnet.Server.Internal
                     }
 
                     // 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, clientConnection, applicationMessage).ConfigureAwait(false);
+                    await InterceptApplicationMessageAsync(undeliveredMessageInterceptor, sender, applicationMessage).ConfigureAwait(false);
                 }
             }
             catch (Exception exception)

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -15,7 +15,7 @@ namespace MQTTnet.Server.Internal
         // Since writing is done by multiple threads (server API or connection thread), we cannot avoid locking
         // completely by swapping references etc.
         readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim();
-        readonly Dictionary<string, Subscription> _subscriptions = new Dictionary<string, Subscription>(4096);
+        readonly Dictionary<string, MqttSubscription> _subscriptions = new Dictionary<string, MqttSubscription>(4096);
 
         readonly MqttClientSession _clientSession;
         readonly IMqttServerOptions _options;
@@ -34,12 +34,12 @@ namespace MQTTnet.Server.Internal
             _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
-        public async Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
+        public async Task<MqttSubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
 
             var retainedApplicationMessages = await _retainedMessagesManager.GetMessagesAsync().ConfigureAwait(false);
-            var result = new SubscribeResult();
+            var result = new MqttSubscribeResult();
 
             // The topic filters are order by its QoS so that the higher QoS will win over a
             // lower one.
@@ -47,18 +47,18 @@ namespace MQTTnet.Server.Internal
             {
                 var interceptorContext = await InterceptSubscribe(originalTopicFilter, cancellationToken).ConfigureAwait(false);
                 var finalTopicFilter = interceptorContext.TopicFilter;
-                var acceptSubscription = interceptorContext.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
+                var processSubscription = interceptorContext.Response.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
 
-                if (string.IsNullOrEmpty(finalTopicFilter.Topic) || !acceptSubscription)
+                if (string.IsNullOrEmpty(finalTopicFilter.Topic) || !processSubscription)
                 {
                     // Return codes is for MQTT < 5 and Reason Code is for MQTT > 5.
                     result.ReturnCodes.Add(MqttSubscribeReturnCode.Failure);
-                    result.ReasonCodes.Add(interceptorContext.ReasonCode);
+                    result.ReasonCodes.Add(interceptorContext.Response.ReasonCode);
                 }
                 else
                 {
-                    result.ReturnCodes.Add(interceptorContext.ReturnCode);
-                    result.ReasonCodes.Add(interceptorContext.ReasonCode);
+                    result.ReturnCodes.Add((MqttSubscribeReturnCode)interceptorContext.Response.ReasonCode);
+                    result.ReasonCodes.Add(interceptorContext.Response.ReasonCode);
                 }
 
                 if (interceptorContext.CloseConnection)
@@ -68,7 +68,7 @@ namespace MQTTnet.Server.Internal
                     result.CloseConnection = true;
                 }
 
-                if (!acceptSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic))
+                if (!processSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic))
                 {
                     continue;
                 }
@@ -78,7 +78,7 @@ namespace MQTTnet.Server.Internal
                     var subscription = CreateSubscription(
                         finalTopicFilter,
                         subscribePacket.Properties?.SubscriptionIdentifier ?? 0,
-                        interceptorContext.ReasonCode);
+                        interceptorContext.Response.ReasonCode);
 
                     await _eventDispatcher
                         .SafeNotifyClientSubscribedTopicAsync(_clientSession.Id, finalTopicFilter)
@@ -91,11 +91,11 @@ namespace MQTTnet.Server.Internal
             return result;
         }
 
-        public async Task<UnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
+        public async Task<MqttUnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
 
-            var result = new UnsubscribeResult();
+            var result = new MqttUnsubscribeResult();
 
             _subscriptionsLock.EnterWriteLock();
             try
@@ -105,9 +105,9 @@ namespace MQTTnet.Server.Internal
                     _subscriptions.TryGetValue(topicFilter, out var existingSubscription);
 
                     var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription, cancellationToken).ConfigureAwait(false);
-                    var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success;
+                    var acceptUnsubscription = interceptorContext.Response.ReasonCode == MqttUnsubscribeReasonCode.Success;
 
-                    result.ReasonCodes.Add(interceptorContext.ReasonCode);
+                    result.ReasonCodes.Add(interceptorContext.Response.ReasonCode);
 
                     if (interceptorContext.CloseConnection)
                     {
@@ -143,7 +143,7 @@ namespace MQTTnet.Server.Internal
         public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel,
             string senderClientId)
         {
-            List<Subscription> subscriptions;
+            List<MqttSubscription> subscriptions;
             _subscriptionsLock.EnterReadLock();
             try
             {
@@ -201,10 +201,9 @@ namespace MQTTnet.Server.Internal
             };
         }
 
-        Subscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier,
-            MqttSubscribeReasonCode reasonCode)
+        MqttSubscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier, MqttSubscribeReasonCode reasonCode)
         {
-            var subscription = new Subscription
+            var subscription = new MqttSubscription
             {
                 Topic = topicFilter.Topic,
                 NoLocal = topicFilter.NoLocal,
@@ -245,7 +244,7 @@ namespace MQTTnet.Server.Internal
         }
 
         static void FilterRetainedApplicationMessages(IList<MqttApplicationMessage> retainedApplicationMessages,
-            Subscription subscription, SubscribeResult subscribeResult)
+            MqttSubscription mqttSubscription, MqttSubscribeResult mqttSubscribeResult)
         {
             for (var i = retainedApplicationMessages.Count - 1; i >= 0; i--)
             {
@@ -255,20 +254,20 @@ namespace MQTTnet.Server.Internal
                     continue;
                 }
 
-                if (subscription.RetainHandling == MqttRetainHandling.DoNotSendOnSubscribe)
+                if (mqttSubscription.RetainHandling == MqttRetainHandling.DoNotSendOnSubscribe)
                 {
                     // This is a MQTT V5+ feature.
                     continue;
                 }
 
-                if (subscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly &&
-                    !subscription.IsNewSubscription)
+                if (mqttSubscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly &&
+                    !mqttSubscription.IsNewSubscription)
                 {
                     // This is a MQTT V5+ feature.
                     continue;
                 }
 
-                if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, subscription.Topic) !=
+                if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, mqttSubscription.Topic) !=
                     MqttTopicFilterCompareResult.IsMatch)
                 {
                     continue;
@@ -278,15 +277,15 @@ namespace MQTTnet.Server.Internal
                 {
                     ApplicationMessage = retainedApplicationMessage,
                     IsRetainedMessage = true,
-                    SubscriptionQualityOfServiceLevel = subscription.GrantedQualityOfServiceLevel
+                    SubscriptionQualityOfServiceLevel = mqttSubscription.GrantedQualityOfServiceLevel
                 };
 
-                if (subscription.Identifier > 0)
+                if (mqttSubscription.Identifier > 0)
                 {
-                    queuedApplicationMessage.SubscriptionIdentifiers = new List<uint> {subscription.Identifier};
+                    queuedApplicationMessage.SubscriptionIdentifiers = new List<uint> {mqttSubscription.Identifier};
                 }
 
-                subscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage);
+                mqttSubscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage);
 
                 retainedApplicationMessages[i] = null;
             }
@@ -305,27 +304,20 @@ namespace MQTTnet.Server.Internal
 
             if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
-                context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS0;
-                context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS0;
+                context.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS0;
             }
             else if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
             {
-                context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS1;
-                context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS1;
+                context.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS1;
             }
             else if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
             {
-                context.ReturnCode = MqttSubscribeReturnCode.SuccessMaximumQoS2;
-                context.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2;
+                context.Response.ReasonCode = MqttSubscribeReasonCode.GrantedQoS2;
             }
-
-            context.DefaultReturnCode = context.ReturnCode;
-            context.DefaultReasonCode = context.ReasonCode;
-
+            
             if (topicFilter.Topic.StartsWith("$share/"))
             {
-                context.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported;
-                context.ReturnCode = MqttSubscribeReturnCode.Failure;
+                context.Response.ReasonCode = MqttSubscribeReasonCode.SharedSubscriptionsNotSupported;
             }
             else
             {
@@ -339,7 +331,7 @@ namespace MQTTnet.Server.Internal
             return context;
         }
 
-        async Task<MqttUnsubscriptionInterceptorContext> InterceptUnsubscribe(string topicFilter, Subscription subscription, CancellationToken cancellationToken)
+        async Task<MqttUnsubscriptionInterceptorContext> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
         {
             var context = new MqttUnsubscriptionInterceptorContext
             {
@@ -349,13 +341,13 @@ namespace MQTTnet.Server.Internal
                 CancellationToken = cancellationToken
             };
 
-            if (subscription == null)
+            if (mqttSubscription == null)
             {
-                context.DefaultReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted;
+                context.Response.ReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted;
             }
             else
             {
-                context.DefaultReasonCode = MqttUnsubscribeReasonCode.Success;
+                context.Response.ReasonCode = MqttUnsubscribeReasonCode.Success;
             }
 
             var interceptor = _options.UnsubscriptionInterceptor;

Source/MQTTnet/Server/Internal/MqttServerEventDispatcher.cs

@@ -2,8 +2,8 @@ using System;
 using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Client.Receiving;
-using MQTTnet.Diagnostics;
 using MQTTnet.Diagnostics.Logger;
+using MQTTnet.Internal;
 using MQTTnet.Packets;
 
 namespace MQTTnet.Server.Internal
@@ -19,6 +19,16 @@ namespace MQTTnet.Server.Internal
             _logger = logger.WithSource(nameof(MqttServerEventDispatcher));
         }
 
+        public AsyncEvent<MqttServerClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<MqttServerClientConnectedEventArgs>();
+        
+        public AsyncEvent<MqttServerClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<MqttServerClientDisconnectedEventArgs>();
+        
+        public AsyncEvent<MqttServerClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<MqttServerClientSubscribedTopicEventArgs>();
+        
+        public AsyncEvent<MqttServerClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<MqttServerClientUnsubscribedTopicEventArgs>();
+        
+        public AsyncEvent<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceivedEvent { get; } = new AsyncEvent<MqttApplicationMessageReceivedEventArgs>();
+        
         public IMqttServerClientConnectedHandler ClientConnectedHandler { get; set; }
 
         public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler { get; set; }
@@ -33,19 +43,21 @@ namespace MQTTnet.Server.Internal
         {
             try
             {
-                var handler = ClientConnectedHandler;
-                if (handler == null)
-                {
-                    return;
-                }
-
-                await handler.HandleClientConnectedAsync(new MqttServerClientConnectedEventArgs
+                var eventArgs = new MqttServerClientConnectedEventArgs
                 {
                     ClientId = connectPacket.ClientId,
                     UserName = connectPacket.Username,
                     ProtocolVersion = channelAdapter.PacketFormatterAdapter.ProtocolVersion,
                     Endpoint = channelAdapter.Endpoint
-                }).ConfigureAwait(false);
+                };
+                
+                var handler = ClientConnectedHandler;
+                if (handler != null)
+                {
+                    await handler.HandleClientConnectedAsync(eventArgs).ConfigureAwait(false);
+                }
+
+                await ClientConnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -57,18 +69,20 @@ namespace MQTTnet.Server.Internal
         {
             try
             {
-                var handler = ClientDisconnectedHandler;
-                if (handler == null)
-                {
-                    return;
-                }
-
-                await handler.HandleClientDisconnectedAsync(new MqttServerClientDisconnectedEventArgs
+                var eventArgs = new MqttServerClientDisconnectedEventArgs
                 {
                     ClientId = clientId,
                     DisconnectType = disconnectType,
                     Endpoint = endpoint
-                }).ConfigureAwait(false);
+                };
+                
+                var handler = ClientDisconnectedHandler;
+                if (handler != null)
+                {
+                    await handler.HandleClientDisconnectedAsync(eventArgs).ConfigureAwait(false);
+                }
+
+                await ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -80,17 +94,19 @@ namespace MQTTnet.Server.Internal
         {
             try
             {
-                var handler = ClientSubscribedTopicHandler;
-                if (handler == null)
-                {
-                    return;
-                }
-
-                await handler.HandleClientSubscribedTopicAsync(new MqttServerClientSubscribedTopicEventArgs
+                var eventArgs = new MqttServerClientSubscribedTopicEventArgs
                 {
                     ClientId = clientId,
                     TopicFilter = topicFilter
-                }).ConfigureAwait(false);
+                };
+                
+                var handler = ClientSubscribedTopicHandler;
+                if (handler != null)
+                {
+                    await handler.HandleClientSubscribedTopicAsync(eventArgs).ConfigureAwait(false);
+                }
+
+                await ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -102,17 +118,19 @@ namespace MQTTnet.Server.Internal
         {
             try
             {
-                var handler = ClientUnsubscribedTopicHandler;
-                if (handler == null)
-                {
-                    return;
-                }
-
-                await handler.HandleClientUnsubscribedTopicAsync(new MqttServerClientUnsubscribedTopicEventArgs
+                var eventArgs = new MqttServerClientUnsubscribedTopicEventArgs
                 {
                     ClientId = clientId,
                     TopicFilter = topicFilter
-                }).ConfigureAwait(false);
+                };
+                
+                var handler = ClientUnsubscribedTopicHandler;
+                if (handler != null)
+                {
+                    await handler.HandleClientUnsubscribedTopicAsync(eventArgs).ConfigureAwait(false);
+                }
+
+                await ClientUnsubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
             catch (Exception exception)
             {
@@ -124,13 +142,15 @@ namespace MQTTnet.Server.Internal
         {
             try
             {
+                var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null);
+                
                 var handler = ApplicationMessageReceivedHandler;
-                if (handler == null)
+                if (handler != null)
                 {
-                    return;
+                    await handler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false);
                 }
 
-                await handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null)).ConfigureAwait(false);
+                await ApplicationMessageReceivedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
             }
             catch (Exception exception)
             {

Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs

@@ -3,7 +3,7 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Server.Internal
 {
-    public sealed class SubscribeResult
+    public sealed class MqttSubscribeResult
     {
         public List<MqttSubscribeReturnCode> ReturnCodes { get; } = new List<MqttSubscribeReturnCode>(128);
 

Source/MQTTnet/Server/Internal/MqttSubscription.cs

@@ -2,7 +2,7 @@
 
 namespace MQTTnet.Server.Internal
 {
-    public sealed class Subscription
+    public sealed class MqttSubscription
     {
         public string Topic { get; set; }
         

Source/MQTTnet/Server/Internal/MqttUnsubscribeResult.cs

@@ -3,7 +3,7 @@ using MQTTnet.Protocol;
 
 namespace MQTTnet.Server.Internal
 {
-    public sealed class UnsubscribeResult
+    public sealed class MqttUnsubscribeResult
     {
         public List<MqttUnsubscribeReasonCode> ReasonCodes { get; } = new List<MqttUnsubscribeReasonCode>(128);
         

Source/MQTTnet/Server/MqttApplicationMessageInterceptorContext.cs

@@ -1,5 +1,5 @@
-using System;
-using System.Collections.Generic;
+using System.Collections.Generic;
+using System.Threading;
 
 namespace MQTTnet.Server
 {
@@ -18,18 +18,21 @@ namespace MQTTnet.Server
         /// </summary>
         public IDictionary<object, object> SessionItems { get; internal set; }
 
-        [Obsolete("Please use ProcessPublish instead.")]
-        public bool AcceptPublish 
-        { 
-            get => ProcessPublish;
-            set => ProcessPublish = value;
-        }
+        /// <summary>
+        /// Gets the response which will be sent to the client via the PUBACK etc. packets.
+        /// </summary>
+        public MqttApplicationMessageResponse Response { get; } = new MqttApplicationMessageResponse();
         
         /// <summary>
         /// Gets or sets whether the publish should be processed internally.
         /// </summary>
         public bool ProcessPublish { get; set; } = true;
-
+        
         public bool CloseConnection { get; set; }
+        
+        /// <summary>
+        /// Gets the cancellation token which can indicate that the client connection gets down.
+        /// </summary>
+        public CancellationToken CancellationToken { get; internal set; }
     }
 }

Source/MQTTnet/Server/MqttApplicationMessageResponse.cs

@@ -0,0 +1,14 @@
+using System.Collections.Generic;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttApplicationMessageResponse
+    {
+        public MqttApplicationMessageResponseReasonCode ReasonCode { get; set; } = MqttApplicationMessageResponseReasonCode.Success;
+        
+        public string ReasonString { get; set; }
+
+        public List<MqttUserProperty> UserProperties { get; } = new List<MqttUserProperty>();
+    }
+}

Source/MQTTnet/Server/MqttApplicationMessageResponseReasonCode.cs

@@ -0,0 +1,17 @@
+namespace MQTTnet.Server
+{
+    public enum MqttApplicationMessageResponseReasonCode
+    {
+
+        Success = 0,
+        //NoMatchingSubscribers = 16,
+        UnspecifiedError = 128,
+        ImplementationSpecificError = 131,
+        NotAuthorized = 135,
+        TopicNameInvalid = 144,
+        //PacketIdentifierInUse = 145,
+        QuotaExceeded = 151,
+        PayloadFormatInvalid = 153
+    
+    }
+}

Source/MQTTnet/Server/MqttServer.cs

@@ -10,16 +10,17 @@ using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
-using MQTTnet.Formatter;
 using MQTTnet.Implementations;
 using MQTTnet.Internal;
 using MQTTnet.Server.Internal;
 
 namespace MQTTnet.Server
 {
-    public class MqttServer : Disposable, IMqttServer
+    public sealed class MqttServer : Disposable, IMqttServer
     {
-        readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
+        readonly AsyncEvent<EventArgs> _startedEvent = new AsyncEvent<EventArgs>();
+        readonly AsyncEvent<EventArgs> _stoppedEvent = new AsyncEvent<EventArgs>();
+        
         readonly MqttServerEventDispatcher _eventDispatcher;
         readonly ICollection<IMqttServerAdapter> _adapters;
         readonly IMqttNetLogger _rootLogger;
@@ -46,38 +47,80 @@ namespace MQTTnet.Server
 
         public IMqttServerStartedHandler StartedHandler { get; set; }
 
+        public event Func<EventArgs, Task> StartedAsync
+        {
+            add => _startedEvent.AddHandler(value);
+            remove => _startedEvent.RemoveHandler(value);
+        }
+
         public IMqttServerStoppedHandler StoppedHandler { get; set; }
 
+        public event Func<EventArgs, Task> StoppedAsync
+        {
+            add => _stoppedEvent.AddHandler(value);
+            remove => _stoppedEvent.RemoveHandler(value);
+        }
+        
         public IMqttServerClientConnectedHandler ClientConnectedHandler
         {
             get => _eventDispatcher.ClientConnectedHandler;
             set => _eventDispatcher.ClientConnectedHandler = value;
         }
 
+        public event Func<MqttServerClientConnectedEventArgs, Task> ClientConnectedAsync
+        {
+            add => _eventDispatcher.ClientConnectedEvent.AddHandler(value);
+            remove => _eventDispatcher.ClientConnectedEvent.RemoveHandler(value);
+        }
+
         public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler
         {
             get => _eventDispatcher.ClientDisconnectedHandler;
             set => _eventDispatcher.ClientDisconnectedHandler = value;
         }
 
+        public event Func<MqttServerClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
+        {
+            add => _eventDispatcher.ClientDisconnectedEvent.AddHandler(value);
+            remove => _eventDispatcher.ClientDisconnectedEvent.RemoveHandler(value);
+        }
+        
         public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler
         {
             get => _eventDispatcher.ClientSubscribedTopicHandler;
             set => _eventDispatcher.ClientSubscribedTopicHandler = value;
         }
 
+        public event Func<MqttServerClientSubscribedTopicEventArgs, Task> ClientSubscribedTopicAsync
+        {
+            add => _eventDispatcher.ClientSubscribedTopicEvent.AddHandler(value);
+            remove => _eventDispatcher.ClientSubscribedTopicEvent.RemoveHandler(value);
+        }
+        
         public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler
         {
             get => _eventDispatcher.ClientUnsubscribedTopicHandler;
             set => _eventDispatcher.ClientUnsubscribedTopicHandler = value;
         }
+        
+        public event Func<MqttServerClientUnsubscribedTopicEventArgs, Task> ClientUnsubscribedTopicAsync
+        {
+            add => _eventDispatcher.ClientUnsubscribedTopicEvent.AddHandler(value);
+            remove => _eventDispatcher.ClientUnsubscribedTopicEvent.RemoveHandler(value);
+        }
 
         public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
         {
             get => _eventDispatcher.ApplicationMessageReceivedHandler;
             set => _eventDispatcher.ApplicationMessageReceivedHandler = value;
         }
-
+        
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => _eventDispatcher.ApplicationMessageReceivedEvent.AddHandler(value);
+            remove => _eventDispatcher.ApplicationMessageReceivedEvent.RemoveHandler(value);
+        }
+        
         public IMqttServerOptions Options { get; private set; }
 
         public Task<IList<IMqttClientStatus>> GetClientStatusAsync()
@@ -139,7 +182,7 @@ namespace MQTTnet.Server
             return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters);
         }
 
-        public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
+        public async Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
         {
             if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
 
@@ -149,10 +192,9 @@ namespace MQTTnet.Server
 
             ThrowIfNotStarted();
 
-            var publishPacket = _packetFactories.Publish.Create(applicationMessage);
-            _clientSessionsManager.DispatchPublishPacket(publishPacket, null);
+            await _clientSessionsManager.DispatchPublishPacket(applicationMessage, null);
 
-            return Task.FromResult(new MqttClientPublishResult());
+            return new MqttClientPublishResult();
         }
 
         public async Task StartAsync(IMqttServerOptions options)
@@ -171,7 +213,7 @@ namespace MQTTnet.Server
             await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
 
             _clientSessionsManager = new MqttClientSessionsManager(Options, _retainedMessagesManager, _eventDispatcher, _rootLogger);
-            _clientSessionsManager.Start(cancellationToken);
+            //_clientSessionsManager.Start(cancellationToken);
 
             _keepAliveMonitor = new MqttServerKeepAliveMonitor(Options, _clientSessionsManager, _rootLogger);
             _keepAliveMonitor.Start(cancellationToken);
@@ -189,6 +231,8 @@ namespace MQTTnet.Server
             {
                 await startedHandler.HandleServerStartedAsync(EventArgs.Empty).ConfigureAwait(false);
             }
+            
+            await _startedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
         }
 
         public async Task StopAsync()
@@ -214,7 +258,7 @@ namespace MQTTnet.Server
             }
             finally
             {
-                _clientSessionsManager?.Dispose();
+                //_clientSessionsManager?.Dispose();
                 _clientSessionsManager = null;
 
                 _cancellationTokenSource?.Dispose();
@@ -228,6 +272,8 @@ namespace MQTTnet.Server
             {
                 await stoppedHandler.HandleServerStoppedAsync(EventArgs.Empty).ConfigureAwait(false);
             }
+
+            await _stoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
         }
 
         protected override void Dispose(bool disposing)

Source/MQTTnet/Server/MqttSubscribeResponse.cs

@@ -0,0 +1,20 @@
+using System.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttSubscribeResponse
+    {
+        /// <summary>
+        /// Gets or sets the reason code which is sent to the client.
+        /// The subscription is skipped when the value is not GrantedQoS_.
+        /// MQTTv5 only.
+        /// </summary>
+        public MqttSubscribeReasonCode ReasonCode { get; set; }
+        
+        public List<MqttUserProperty> UserProperties { get; } = new List<MqttUserProperty>();
+        
+        public string ReasonString { get; set; }
+    }
+}

Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs

@@ -1,7 +1,5 @@
-using System;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Threading;
-using MQTTnet.Protocol;
 using MQTTnet.Server.Status;
 
 namespace MQTTnet.Server
@@ -24,48 +22,14 @@ namespace MQTTnet.Server
         /// Gets or sets a key/value collection that can be used to share data within the scope of this session.
         /// </summary>
         public IDictionary<object, object> SessionItems { get; internal set; }
-
-        // Will be removed together with "AcceptSubscription". It only stores the default value when setting "AcceptSubscription" to true.
-        internal MqttSubscribeReasonCode DefaultReasonCode { get; set; }
         
-        internal MqttSubscribeReturnCode DefaultReturnCode { get; set; }
-
-        [Obsolete("Please use a proper value for 'ReasonCode' instead. This property will be removed in the future.")]
-        public bool AcceptSubscription
-        {
-            get => ReasonCode == MqttSubscribeReasonCode.GrantedQoS0 && ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
-            set
-            {
-                if (value)
-                {
-                    ReasonCode = DefaultReasonCode;
-                    ReturnCode = DefaultReturnCode;
-                }
-                else
-                {
-                    ReturnCode = MqttSubscribeReturnCode.Failure;
-                    ReasonCode = MqttSubscribeReasonCode.UnspecifiedError;
-                }
-            }
-        }
-
         /// <summary>
         /// Gets or sets whether the broker should create an internal subscription for the client.
         /// The broker can also avoid this and return "success" to the client.
         /// This feature allows using the MQTT Broker as the Frontend and another system as the backend.
         /// </summary>
         public bool ProcessSubscription { get; set; } = true;
-
-        /// <summary>
-        /// Gets or sets the reason code which is sent to the client.
-        /// The subscription is skipped when the value is not GrantedQoS_.
-        /// MQTTv5 only.
-        /// </summary>
-        public MqttSubscribeReasonCode ReasonCode { get; set; }
-
-        // MQTT < 5 only!
-        internal MqttSubscribeReturnCode ReturnCode { get; set; }
-
+       
         /// <summary>
         /// Gets or sets whether the broker should close the client connection.
         /// </summary>
@@ -75,6 +39,11 @@ namespace MQTTnet.Server
         /// Gets the current client session.
         /// </summary>
         public IMqttSessionStatus Session { get; internal set; }
+
+        /// <summary>
+        /// Gets the response which will be sent to the client via the SUBACK packet.
+        /// </summary>
+        public MqttSubscribeResponse Response { get; } = new MqttSubscribeResponse();
         
         /// <summary>
         /// Gets the cancellation token which can indicate that the client connection gets down.

Source/MQTTnet/Server/MqttUnsubscribeResponse.cs

@@ -0,0 +1,19 @@
+using System.Collections.Generic;
+using MQTTnet.Packets;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server
+{
+    public sealed class MqttUnsubscribeResponse
+    {
+        /// <summary>
+        /// Gets or sets the reason code which is sent to the client.
+        /// MQTTv5 only.
+        /// </summary>
+        public MqttUnsubscribeReasonCode ReasonCode { get; set; }
+        
+        public List<MqttUserProperty> UserProperties { get; } = new List<MqttUserProperty>();
+        
+        public string ReasonString { get; set; }
+    }
+}

Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs

@@ -1,7 +1,5 @@
-using System;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Threading;
-using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
@@ -25,21 +23,10 @@ namespace MQTTnet.Server
         /// </summary>
         public IDictionary<object, object> SessionItems { get; internal set; }
 
-        [Obsolete("Please use ReasonCode instead.")]
-        public bool AcceptUnsubscription
-        {
-            get => ReasonCode == MqttUnsubscribeReasonCode.Success;
-            set => ReasonCode = DefaultReasonCode;
-        }
-
-        // Gets removed together with "AcceptUnsubscription".
-        internal MqttUnsubscribeReasonCode DefaultReasonCode { get; set; }
-        
         /// <summary>
-        /// Gets or sets the reason code which will be sent to the client.
-        /// This value is not transmitted when using MQTT v3.1.1.
+        /// Gets the response which will be sent to the client via the UNSUBACK pocket.
         /// </summary>
-        public MqttUnsubscribeReasonCode ReasonCode { get; set; }
+        public MqttUnsubscribeResponse Response { get; } = new MqttUnsubscribeResponse();
         
         /// <summary>
         /// Gets or sets whether the broker should remove an internal subscription for the client.

Tests/MQTTnet.AspNetCore.Tests/ReaderExtensionsTest.cs

@@ -15,7 +15,7 @@ namespace MQTTnet.AspNetCore.Tests
         {
             var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311);
 
-            var buffer = serializer.Encode(new MqttPublishPacket() {Topic = "a", Payload = new byte[5]});
+            var buffer = serializer.Encode(new MqttPublishPacket {Topic = "a", Payload = new byte[5]}).ToArray();
 
             var sequence = new ReadOnlySequence<byte>(buffer.Array, buffer.Offset, buffer.Count);
 

Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs

@@ -28,7 +28,7 @@ namespace MQTTnet.Benchmarks
 
             var serializer = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311);
             
-            var serializedPacket = Join(serializer.Encode(_packet));
+            var serializedPacket = Join(serializer.Encode(_packet).ToArray());
 
             _iterations = 10000;
 

Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -31,7 +31,7 @@ namespace MQTTnet.Benchmarks
             };
 
             _serializer = new MqttV311PacketFormatter(new MqttPacketWriter());
-            _serializedPacket = _serializer.Encode(_packet);
+            _serializedPacket = _serializer.Encode(_packet).ToArray();
         }
 
         [Benchmark]

Tests/MQTTnet.Core.Tests/Client/LowLevelMqttClient_Tests.cs

@@ -84,7 +84,7 @@ namespace MQTTnet.Tests.Client
                 await lowLevelClient.DisconnectAsync(CancellationToken.None).ConfigureAwait(false);
 
                 Assert.IsNotNull(receivedPacket);
-                Assert.AreEqual(MqttSubscribeReturnCode.SuccessMaximumQoS0, receivedPacket.ReturnCodes[0]);
+                Assert.AreEqual(MqttSubscribeReasonCode.GrantedQoS0, receivedPacket.ReasonCodes[0]);
             }
         }
 

Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs

@@ -36,18 +36,36 @@ namespace MQTTnet.Tests.Mockups
             set => Implementation.ConnectedHandler = value;
         }
 
+        public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync
+        {
+            add => Implementation.ConnectedAsync += value;
+            remove => Implementation.ConnectedAsync -= value;
+        }
+
         public IMqttClientDisconnectedHandler DisconnectedHandler
         {
             get => Implementation.DisconnectedHandler;
             set => Implementation.DisconnectedHandler = value;
         }
 
+        public event Func<MqttClientDisconnectedEventArgs, Task> DisconnectedAsync
+        {
+            add => Implementation.DisconnectedAsync += value;
+            remove => Implementation.DisconnectedAsync -= value;
+        }
+
         public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
         {
             get => Implementation.ApplicationMessageReceivedHandler;
             set => Implementation.ApplicationMessageReceivedHandler = value;
         }
 
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => Implementation.ApplicationMessageReceivedAsync += value;
+            remove => Implementation.ApplicationMessageReceivedAsync -= value;
+        }
+
         public Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken)
         {
             if (TestContext != null)

Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs

@@ -69,6 +69,12 @@ namespace MQTTnet.Tests.Mockups
             set => Implementation.ApplicationMessageReceivedHandler = value;
         }
 
+        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync
+        {
+            add => Implementation.ApplicationMessageReceivedAsync += value;
+            remove => Implementation.ApplicationMessageReceivedAsync -= value;
+        }
+
         public Task ClearRetainedApplicationMessagesAsync()
         {
             return Implementation.ClearRetainedApplicationMessagesAsync();

Tests/MQTTnet.Core.Tests/MqttPacketSerializer_Tests.cs

@@ -513,10 +513,10 @@ namespace MQTTnet.Tests
                 PacketIdentifier = 123
             };
 
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS0);
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1);
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS2);
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.Failure);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS0);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS1);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS2);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.UnspecifiedError);
 
             SerializeAndCompare(p, "kAYAewABAoA=");
         }
@@ -529,10 +529,10 @@ namespace MQTTnet.Tests
                 PacketIdentifier = 123
             };
 
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS0);
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS1);
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.SuccessMaximumQoS2);
-            p.ReturnCodes.Add(MqttSubscribeReturnCode.Failure);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS0);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS1);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.GrantedQoS2);
+            p.ReasonCodes.Add(MqttSubscribeReasonCode.UnspecifiedError);
 
             DeserializeAndCompare(p, "kAYAewABAoA=");
         }
@@ -596,7 +596,7 @@ namespace MQTTnet.Tests
 
         byte[] Serialize(MqttBasePacket packet, MqttProtocolVersion protocolVersion)
         {
-            return MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, WriterFactory()).Encode(packet).ToArray();
+            return MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, WriterFactory()).Encode(packet).ToArray().ToArray();
         }
 
         protected virtual IMqttPacketWriter WriterFactory()
@@ -616,7 +616,7 @@ namespace MQTTnet.Tests
             var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
             var buffer1 = serializer.Encode(packet);
 
-            using (var headerStream = new MemoryStream(buffer1.ToArray()))
+            using (var headerStream = new MemoryStream(buffer1.ToArray().ToArray()))
             {
                 var channel = new TestMqttChannel(headerStream);
                 var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, writer), null, new MqttNetEventLogger());
@@ -624,20 +624,7 @@ namespace MQTTnet.Tests
 
                 var buffer2 = serializer.Encode(receivedPacket);
 
-                Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer2.ToArray()));
-
-                //adapter.ReceivePacketAsync(CancellationToken.None);
-                //var fixedHeader = new byte[2];
-                //var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult().FixedHeader;
-
-                //using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength))
-                //{
-                //    var reader = ReaderFactory(bodyStream.ToArray());
-                //    var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0));
-                //    var buffer2 = serializer.Encode(deserializedPacket);
-
-                //    Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));
-                //}
+                Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer2.ToArray().ToArray()));
             }
         }
 
@@ -648,26 +635,9 @@ namespace MQTTnet.Tests
             var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
             var buffer = serializer.Encode(packet);
             
-            var channel = new TestMqttChannel(buffer.ToArray());
+            var channel = new TestMqttChannel(buffer.ToArray().ToArray());
             var adapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(protocolVersion, writer), null, new MqttNetEventLogger());
             return (TPacket)adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
-
-            //using (var headerStream = new MemoryStream(buffer1.ToArray()))
-            //{
-
-
-
-
-            //    //var fixedHeader = new byte[2];
-
-            //    //var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult().FixedHeader;
-
-            //    //using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, (int)header.RemainingLength))
-            //    //{
-            //    //    var reader = ReaderFactory(bodyStream.ToArray());
-            //    //    return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0));
-            //    //}
-            //}
         }
 
         MqttProtocolVersion DeserializeAndDetectVersion(MqttPacketFormatterAdapter packetFormatterAdapter, byte[] buffer)
@@ -677,24 +647,6 @@ namespace MQTTnet.Tests
 
             adapter.ReceivePacketAsync(CancellationToken.None).GetAwaiter().GetResult();
             return packetFormatterAdapter.ProtocolVersion;
-
-            //using (var headerStream = new MemoryStream(buffer))
-            //{
-
-
-
-
-            //    //var fixedHeader = new byte[2];
-            //    //var header = new MqttPacketReader(channel).ReadFixedHeaderAsync(fixedHeader, CancellationToken.None).GetAwaiter().GetResult().FixedHeader;
-
-            //    //using (var bodyStream = new MemoryStream(buffer, (int)headerStream.Position, (int)header.RemainingLength))
-            //    //{
-            //    //    var reader = ReaderFactory(bodyStream.ToArray());
-            //    //    var packet = new ReceivedMqttPacket(header.Flags, reader, 0);
-            //    //    packetFormatterAdapter.DetectProtocolVersion(packet);
-            //    //    return adapter.ProtocolVersion;
-            //    //}
-            //}
         }
     }
 }

Tests/MQTTnet.Core.Tests/Server/General.cs

@@ -878,7 +878,7 @@ namespace MQTTnet.Tests.Server
                     // This should lead to no subscriptions for "n" at all. So also no sending of retained messages.
                     if (c.TopicFilter.Topic == "n")
                     {
-                        c.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError;
+                        c.Response.ReasonCode = MqttSubscribeReasonCode.UnspecifiedError;
                     }
                 }));
 

Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs

@@ -34,7 +34,7 @@ namespace MQTTnet.Tests.Server
                         {
                             if (context.SessionItems["can_subscribe_x"] as bool? == false)
                             {
-                                context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
+                                context.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
                             }
                         }
                     })

Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs

@@ -265,7 +265,7 @@ namespace MQTTnet.Tests.Server
                     {
                         if (c.TopicFilter.Topic == "not_allowed_topic")
                         {
-                            c.ReasonCode = MqttSubscribeReasonCode.TopicFilterInvalid;
+                            c.Response.ReasonCode = MqttSubscribeReasonCode.TopicFilterInvalid;
                         }
                     }));
                 

Tests/MQTTnet.TestApp.NetCore/ServerTest.cs

@@ -73,12 +73,12 @@ namespace MQTTnet.TestApp.NetCore
                     {
                         if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
                         {
-                            context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
+                            context.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
                         }
 
                         if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
                         {
-                            context.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
+                            context.Response.ReasonCode = MqttSubscribeReasonCode.ImplementationSpecificError;
                             context.CloseConnection = true;
                         }
                     })