Extend interceptors.

Source/MQTTnet/Formatter/IMqttDataConverter.cs

@@ -46,7 +46,7 @@ namespace MQTTnet.Formatter
 
         MqttUnsubscribePacket CreateUnsubscribePacket(MqttClientUnsubscribeOptions options);
 
-        MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes);
+        MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult unsubscribeResult);
 
         MqttDisconnectPacket CreateDisconnectPacket(MqttClientDisconnectOptions options);
     }

Source/MQTTnet/Formatter/V3/MqttV310DataConverter.cs

@@ -9,7 +9,6 @@ using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 using System;
-using System.Collections.Generic;
 using System.Linq;
 using MQTTnet.Server.Internal;
 
@@ -201,9 +200,11 @@ namespace MQTTnet.Formatter.V3
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            var subscribePacket = new MqttSubscribePacket();
-            subscribePacket.TopicFilters.AddRange(options.TopicFilters);
-
+            var subscribePacket = new MqttSubscribePacket
+            {
+                TopicFilters = options.TopicFilters
+            };
+            
             return subscribePacket;
         }
 
@@ -214,11 +215,10 @@ namespace MQTTnet.Formatter.V3
 
             var subackPacket = new MqttSubAckPacket
             {
-                PacketIdentifier = subscribePacket.PacketIdentifier
+                PacketIdentifier = subscribePacket.PacketIdentifier,
+                ReturnCodes = subscribeResult.ReturnCodes
             };
-
-            subackPacket.ReturnCodes.AddRange(subscribeResult.ReturnCodes);
-
+            
             return subackPacket;
         }
 
@@ -226,21 +226,22 @@ namespace MQTTnet.Formatter.V3
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
-            var unsubscribePacket = new MqttUnsubscribePacket();
-            unsubscribePacket.TopicFilters.AddRange(options.TopicFilters);
+            var unsubscribePacket = new MqttUnsubscribePacket
+            {
+                TopicFilters = options.TopicFilters
+            };
 
             return unsubscribePacket;
         }
 
-        public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes)
+        public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, UnsubscribeResult _)
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
-            if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes));
-
+            
+            // The UnsubscribeResult is not supported in protocol version 3.1.1!
             return new MqttUnsubAckPacket
             {
-                PacketIdentifier = unsubscribePacket.PacketIdentifier,
-                ReasonCodes = reasonCodes
+                PacketIdentifier = unsubscribePacket.PacketIdentifier
             };
         }
 

Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs

@@ -49,47 +49,51 @@ namespace MQTTnet.Formatter.V5
             return packet;
         }
 
-        public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket,
+            MqttApplicationMessageReceivedReasonCode reasonCode)
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
             return new MqttPubAckPacket
             {
                 PacketIdentifier = publishPacket.PacketIdentifier,
-                ReasonCode = (MqttPubAckReasonCode)(int)reasonCode
+                ReasonCode = (MqttPubAckReasonCode) (int) reasonCode
             };
         }
 
-        public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        public MqttPubRecPacket CreatePubRecPacket(MqttPublishPacket publishPacket,
+            MqttApplicationMessageReceivedReasonCode reasonCode)
         {
             if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
 
             return new MqttPubRecPacket
             {
                 PacketIdentifier = publishPacket.PacketIdentifier,
-                ReasonCode = (MqttPubRecReasonCode)(int)reasonCode
+                ReasonCode = (MqttPubRecReasonCode) (int) reasonCode
             };
         }
 
-        public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        public MqttPubCompPacket CreatePubCompPacket(MqttPubRelPacket pubRelPacket,
+            MqttApplicationMessageReceivedReasonCode reasonCode)
         {
             if (pubRelPacket == null) throw new ArgumentNullException(nameof(pubRelPacket));
 
             return new MqttPubCompPacket
             {
                 PacketIdentifier = pubRelPacket.PacketIdentifier,
-                ReasonCode = (MqttPubCompReasonCode)(int)reasonCode
+                ReasonCode = (MqttPubCompReasonCode) (int) reasonCode
             };
         }
 
-        public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket, MqttApplicationMessageReceivedReasonCode reasonCode)
+        public MqttPubRelPacket CreatePubRelPacket(MqttPubRecPacket pubRecPacket,
+            MqttApplicationMessageReceivedReasonCode reasonCode)
         {
             if (pubRecPacket == null) throw new ArgumentNullException(nameof(pubRecPacket));
 
             return new MqttPubRelPacket
             {
                 PacketIdentifier = pubRecPacket.PacketIdentifier,
-                ReasonCode = (MqttPubRelReasonCode)(int)reasonCode
+                ReasonCode = (MqttPubRelReasonCode) (int) reasonCode
             };
         }
 
@@ -122,7 +126,7 @@ namespace MQTTnet.Formatter.V5
             return new MqttClientConnectResult
             {
                 IsSessionPresent = connAckPacket.IsSessionPresent,
-                ResultCode = (MqttClientConnectResultCode)(int)connAckPacket.ReasonCode,
+                ResultCode = (MqttClientConnectResultCode) (int) connAckPacket.ReasonCode,
                 WildcardSubscriptionAvailable = connAckPacket.Properties?.WildcardSubscriptionAvailable ?? true,
                 RetainAvailable = connAckPacket.Properties?.RetainAvailable ?? true,
                 AssignedClientIdentifier = connAckPacket.Properties?.AssignedClientIdentifier,
@@ -143,7 +147,8 @@ namespace MQTTnet.Formatter.V5
             };
         }
 
-        public MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage, IMqttClientOptions options)
+        public MqttConnectPacket CreateConnectPacket(MqttApplicationMessage willApplicationMessage,
+            IMqttClientOptions options)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
 
@@ -153,7 +158,7 @@ namespace MQTTnet.Formatter.V5
                 Username = options.Credentials?.Username,
                 Password = options.Credentials?.Password,
                 CleanSession = options.CleanSession,
-                KeepAlivePeriod = (ushort)options.KeepAlivePeriod.TotalSeconds,
+                KeepAlivePeriod = (ushort) options.KeepAlivePeriod.TotalSeconds,
                 WillMessage = willApplicationMessage,
                 Properties = new MqttConnectPacketProperties
                 {
@@ -185,7 +190,7 @@ namespace MQTTnet.Formatter.V5
                     SharedSubscriptionAvailable = false,
                     TopicAliasMaximum = ushort.MaxValue,
                     WildcardSubscriptionAvailable = true,
-                    
+
                     UserProperties = connectionValidatorContext.ResponseUserProperties,
                     AuthenticationMethod = connectionValidatorContext.AuthenticationMethod,
                     AuthenticationData = connectionValidatorContext.ResponseAuthenticationData,
@@ -196,38 +201,43 @@ namespace MQTTnet.Formatter.V5
             };
         }
 
-        public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket, MqttSubAckPacket subAckPacket)
+        public MqttClientSubscribeResult CreateClientSubscribeResult(MqttSubscribePacket subscribePacket,
+            MqttSubAckPacket subAckPacket)
         {
             if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
             if (subAckPacket == null) throw new ArgumentNullException(nameof(subAckPacket));
 
             if (subAckPacket.ReasonCodes.Count != subscribePacket.TopicFilters.Count)
             {
-                throw new MqttProtocolViolationException("The reason codes are not matching the topic filters [MQTT-3.9.3-1].");
+                throw new MqttProtocolViolationException(
+                    "The reason codes are not matching the topic filters [MQTT-3.9.3-1].");
             }
 
             var result = new MqttClientSubscribeResult();
 
             result.Items.AddRange(subscribePacket.TopicFilters.Select((t, i) =>
-                new MqttClientSubscribeResultItem(t, (MqttClientSubscribeResultCode)subAckPacket.ReasonCodes[i])));
+                new MqttClientSubscribeResultItem(t, (MqttClientSubscribeResultCode) subAckPacket.ReasonCodes[i])));
 
             return result;
         }
 
-        public MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket, MqttUnsubAckPacket unsubAckPacket)
+        public MqttClientUnsubscribeResult CreateClientUnsubscribeResult(MqttUnsubscribePacket unsubscribePacket,
+            MqttUnsubAckPacket unsubAckPacket)
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
             if (unsubAckPacket == null) throw new ArgumentNullException(nameof(unsubAckPacket));
 
             if (unsubAckPacket.ReasonCodes.Count != unsubscribePacket.TopicFilters.Count)
             {
-                throw new MqttProtocolViolationException("The return codes are not matching the topic filters [MQTT-3.9.3-1].");
+                throw new MqttProtocolViolationException(
+                    "The return codes are not matching the topic filters [MQTT-3.9.3-1].");
             }
 
             var result = new MqttClientUnsubscribeResult();
 
             result.Items.AddRange(unsubscribePacket.TopicFilters.Select((t, i) =>
-                new MqttClientUnsubscribeResultItem(t, (MqttClientUnsubscribeResultCode)unsubAckPacket.ReasonCodes[i])));
+                new MqttClientUnsubscribeResultItem(t,
+                    (MqttClientUnsubscribeResultCode) unsubAckPacket.ReasonCodes[i])));
 
             return result;
         }
@@ -255,11 +265,10 @@ namespace MQTTnet.Formatter.V5
 
             var subackPacket = new MqttSubAckPacket
             {
-                PacketIdentifier = subscribePacket.PacketIdentifier
+                PacketIdentifier = subscribePacket.PacketIdentifier,
+                ReasonCodes = subscribeResult.ReasonCodes
             };
 
-            subackPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes);
-
             return subackPacket;
         }
 
@@ -269,24 +278,26 @@ namespace MQTTnet.Formatter.V5
 
             var packet = new MqttUnsubscribePacket
             {
-                Properties = new MqttUnsubscribePacketProperties()
+                TopicFilters = options.TopicFilters,
+                Properties = new MqttUnsubscribePacketProperties
+                {
+                    UserProperties = options.UserProperties
+                }
             };
 
-            packet.TopicFilters.AddRange(options.TopicFilters);
-            packet.Properties.UserProperties = options.UserProperties;
-
             return packet;
         }
 
-        public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket, List<MqttUnsubscribeReasonCode> reasonCodes)
+        public MqttUnsubAckPacket CreateUnsubAckPacket(MqttUnsubscribePacket unsubscribePacket,
+            UnsubscribeResult unsubscribeResult)
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
-            if (reasonCodes == null) throw new ArgumentNullException(nameof(reasonCodes));
+            if (unsubscribeResult == null) throw new ArgumentNullException(nameof(unsubscribeResult));
 
             return new MqttUnsubAckPacket
             {
                 PacketIdentifier = unsubscribePacket.PacketIdentifier,
-                ReasonCodes = reasonCodes
+                ReasonCodes = unsubscribeResult.ReasonCodes
             };
         }
 
@@ -300,7 +311,7 @@ namespace MQTTnet.Formatter.V5
             }
             else
             {
-                packet.ReasonCode = (MqttDisconnectReasonCode)options.ReasonCode;
+                packet.ReasonCode = (MqttDisconnectReasonCode) options.ReasonCode;
             }
 
             return packet;
@@ -319,7 +330,7 @@ namespace MQTTnet.Formatter.V5
             {
                 // QoS 0 has no response. So we treat it as a success always.
                 // Both enums have the same values. So it can be easily converted.
-                result.ReasonCode = (MqttClientPublishReasonCode)(int)(pubAckPacket.ReasonCode ?? 0);
+                result.ReasonCode = (MqttClientPublishReasonCode) (int) (pubAckPacket.ReasonCode ?? 0);
 
                 result.PacketIdentifier = pubAckPacket.PacketIdentifier;
             }
@@ -327,7 +338,8 @@ namespace MQTTnet.Formatter.V5
             return result;
         }
 
-        public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket, MqttPubCompPacket pubCompPacket)
+        public MqttClientPublishResult CreateClientPublishResult(MqttPubRecPacket pubRecPacket,
+            MqttPubCompPacket pubCompPacket)
         {
             if (pubRecPacket == null || pubCompPacket == null)
             {
@@ -360,7 +372,7 @@ namespace MQTTnet.Formatter.V5
             if (pubRecPacket.ReasonCode.HasValue)
             {
                 // Both enums share the same values.
-                result.ReasonCode = (MqttClientPublishReasonCode)(pubRecPacket.ReasonCode ?? 0);
+                result.ReasonCode = (MqttClientPublishReasonCode) (pubRecPacket.ReasonCode ?? 0);
             }
 
             return result;

Source/MQTTnet/Internal/CrossPlatformPromise.cs

@@ -0,0 +1,68 @@
+using System;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Internal
+{
+    public sealed class CrossPlatformPromise<TResult>
+    {
+        readonly TaskCompletionSource<TResult> _taskCompletionSource;
+
+        public CrossPlatformPromise()
+        {
+#if NET452
+            _taskCompletionSource = new TaskCompletionSource<TResult>();
+#else
+            _taskCompletionSource = new TaskCompletionSource<TResult>(TaskCreationOptions.RunContinuationsAsynchronously);
+#endif
+        }
+
+        public Task<TResult> Task => _taskCompletionSource.Task;
+
+        public bool TrySetResult(TResult result)
+        {
+#if NET452
+            // To prevent deadlocks it is required to call the _TrySetResult_ method
+            // from a new thread because the awaiting code will not(!) be executed in
+            // a new thread automatically (due to await). Furthermore _this_ thread will
+            // do it. But _this_ thread is also reading incoming packets -> deadlock.
+            // NET452 does not support RunContinuationsAsynchronously
+            System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetResult(result));
+            return true;
+#else
+            return _taskCompletionSource.TrySetResult(result);
+#endif
+        }
+
+        public bool TrySetException(Exception exception)
+        {
+            if (exception == null) throw new ArgumentNullException(nameof(exception));
+            
+#if NET452
+            // To prevent deadlocks it is required to call the _TrySetException_ method
+            // from a new thread because the awaiting code will not(!) be executed in
+            // a new thread automatically (due to await). Furthermore _this_ thread will
+            // do it. But _this_ thread is also reading incoming packets -> deadlock.
+            // NET452 does not support RunContinuationsAsynchronously
+            System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetException(exception));
+            return true;
+#else
+            return _taskCompletionSource.TrySetException(exception);
+#endif
+        }
+        
+        public bool TrySetCanceled()
+        {
+#if NET452
+            // To prevent deadlocks it is required to call the _TrySetCanceled_ method
+            // from a new thread because the awaiting code will not(!) be executed in
+            // a new thread automatically (due to await). Furthermore _this_ thread will
+            // do it. But _this_ thread is also reading incoming packets -> deadlock.
+            // NET452 does not support RunContinuationsAsynchronously
+            System.Threading.Tasks.Task.Run(() => _taskCompletionSource.TrySetCanceled());
+            return true;
+#else
+            return _taskCompletionSource.TrySetCanceled();
+#endif
+        }
+    }
+}

Source/MQTTnet/Internal/MqttPacketBus.cs

@@ -0,0 +1,113 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Internal
+{
+    public sealed class MqttPacketBus : IDisposable
+    {
+        readonly object _syncRoot = new object();
+
+        readonly Queue<MqttPacketBusItem>[] _partitions = new Queue<MqttPacketBusItem>[]
+        {
+            new Queue<MqttPacketBusItem>(4096),
+            new Queue<MqttPacketBusItem>(1028),
+            new Queue<MqttPacketBusItem>(128)
+        };
+
+        readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);
+
+        int _activePartition = (int) MqttPacketBusPartition.Health;
+
+        public int PacketsCount
+        {
+            get
+            {
+                lock (_syncRoot)
+                {
+                    return _partitions.Sum(p => p.Count);
+                }
+            }
+        }
+
+        public int DataPacketsCount()
+        {
+            lock (_syncRoot)
+            {
+                return _partitions[(int)MqttPacketBusPartition.Data].Count;
+            }
+        }
+
+        public List<MqttBasePacket> ExportPackets(MqttPacketBusPartition partition)
+        {
+            lock (_syncRoot)
+            {
+                return _partitions[(int) partition].Select(i => i.Packet).ToList();
+            }
+        }
+
+        public MqttPacketBusItem Enqueue(MqttBasePacket packet, MqttPacketBusPartition partition)
+        {
+            var busItem = new MqttPacketBusItem(packet);
+            Enqueue(busItem, partition);
+            return busItem;
+        }
+        
+        public void Enqueue(MqttPacketBusItem item, MqttPacketBusPartition partition)
+        {
+            if (item == null) throw new ArgumentNullException(nameof(item));
+
+            lock (_syncRoot)
+            {
+                _partitions[(int) partition].Enqueue(item);
+            }
+
+            _semaphore.Release();
+        }
+
+        void MoveActivePartition()
+        {
+            if (_activePartition >= _partitions.Length - 1)
+            {
+                _activePartition = 0;
+            }
+            else
+            {
+                _activePartition++;
+            }
+        }
+
+        public async Task<MqttPacketBusItem> DequeueAsync(CancellationToken cancellationToken)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                lock (_syncRoot)
+                {
+                    for (var i = 0; i < 3; i++)
+                    {
+                        MoveActivePartition();
+
+                        if (_partitions[_activePartition].Count > 0)
+                        {
+                            return _partitions[_activePartition].Dequeue();
+                        }
+                    }
+                }
+
+                // No partition contains data so that we have to wait and put
+                // the worker back to the thread pool.
+                await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+            }
+
+            throw new InvalidOperationException("MqttPacketBus is broken.");
+        }
+
+        public void Dispose()
+        {
+            _semaphore?.Dispose();
+        }
+    }
+}

Source/MQTTnet/Internal/MqttPacketBusItem.cs

@@ -0,0 +1,45 @@
+using System;
+using System.Threading.Tasks;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Internal
+{
+    public sealed class MqttPacketBusItem
+    {
+        readonly CrossPlatformPromise<int> _promise = new CrossPlatformPromise<int>();
+
+        public MqttPacketBusItem(MqttBasePacket packet)
+        {
+            Packet = packet ?? throw new ArgumentNullException(nameof(packet));
+        }
+        
+        public MqttBasePacket Packet { get; }
+
+        public event EventHandler Delivered;
+
+        public Task WaitForDeliveryAsync()
+        {
+            return _promise.Task;
+        }
+        
+        public void MarkAsDelivered()
+        {
+            if (_promise.TrySetResult(0))
+            {
+                Delivered?.Invoke(this, EventArgs.Empty);
+            }
+        }
+
+        public void MarkAsFailed(Exception exception)
+        {
+            if (exception == null) throw new ArgumentNullException(nameof(exception));
+            
+            _promise.TrySetException(exception);
+        }
+
+        public void MarkAsCancelled()
+        {
+            _promise.TrySetCanceled();
+        }
+    }
+}

Source/MQTTnet/Internal/MqttPacketBusPartition.cs

@@ -0,0 +1,11 @@
+namespace MQTTnet.Internal
+{
+    public enum MqttPacketBusPartition
+    {
+        Data,
+        
+        Control,
+        
+        Health
+    }
+}

Source/MQTTnet/Packets/MqttSubAckPacket.cs

@@ -8,15 +8,20 @@ namespace MQTTnet.Packets
     {
         public ushort PacketIdentifier { get; set; }
 
+        /// <summary>
+        /// Only available in MQTT v3.1.1.
+        /// </summary>
         public List<MqttSubscribeReturnCode> ReturnCodes { get; set; } = new List<MqttSubscribeReturnCode>();
+        
+        /// <summary>
+        /// Added in MQTTv5.
+        /// </summary>
+        public List<MqttSubscribeReasonCode> ReasonCodes { get; set; } = new List<MqttSubscribeReasonCode>();
 
-        #region Added in MQTTv5.0.0
-
-        public List<MqttSubscribeReasonCode> ReasonCodes { get; } = new List<MqttSubscribeReasonCode>();
-
-        public MqttSubAckPacketProperties Properties { get; set; }
-
-        #endregion
+        /// <summary>
+        /// Added in MQTTv5.
+        /// </summary>
+        public MqttSubAckPacketProperties Properties { get; set; } = new MqttSubAckPacketProperties();
 
         public override string ToString()
         {

Source/MQTTnet/Packets/MqttUnsubAckPacket.cs

@@ -8,14 +8,16 @@ namespace MQTTnet.Packets
     {
         public ushort PacketIdentifier { get; set; }
 
-        #region Added in MQTTv5
-
-        public MqttUnsubAckPacketProperties Properties { get; set; }
+        /// <summary>
+        /// Added in MQTT V5.
+        /// </summary>
+        public MqttUnsubAckPacketProperties Properties { get; set; } = new MqttUnsubAckPacketProperties();
 
+        /// <summary>
+        /// Added in MQTT V5.
+        /// </summary>
         public List<MqttUnsubscribeReasonCode> ReasonCodes { get; set; } = new List<MqttUnsubscribeReasonCode>();
 
-        #endregion
-
         public override string ToString()
         {
             var reasonCodesText = string.Join(",", ReasonCodes.Select(f => f.ToString()));

Source/MQTTnet/Packets/MqttUnsubscribePacket.cs

@@ -8,12 +8,11 @@ namespace MQTTnet.Packets
 
         public List<string> TopicFilters { get; set; } = new List<string>();
 
-        #region Added in MQTTv5
-
-        public MqttUnsubscribePacketProperties Properties { get; set; }
-
-        #endregion
-
+        /// <summary>
+        /// Added in MQTTv5.
+        /// </summary>
+        public MqttUnsubscribePacketProperties Properties { get; set; } = new MqttUnsubscribePacketProperties();
+        
         public override string ToString()
         {
             var topicFiltersText = string.Join(",", TopicFilters);

Source/MQTTnet/Server/Internal/MqttClientConnection.cs

@@ -379,10 +379,15 @@ namespace MQTTnet.Server.Internal
 
         async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken)
         {
-            var reasonCodes = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket).ConfigureAwait(false);
-            var unsubAckPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, reasonCodes);
+            var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket).ConfigureAwait(false);
+            var unsubAckPacket = _channelAdapter.PacketFormatterAdapter.DataConverter.CreateUnsubAckPacket(unsubscribePacket, unsubscribeResult);
 
             await SendPacketAsync(unsubAckPacket, cancellationToken).ConfigureAwait(false);
+            
+            if (unsubscribeResult.CloseConnection)
+            {
+                StopInternal();
+            }
         }
 
         Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken)

Source/MQTTnet/Server/Internal/MqttClientSession.cs

@@ -22,6 +22,8 @@ namespace MQTTnet.Server.Internal
             ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
         }
 
+        public event EventHandler Deleted;
+        
         public string ClientId { get; }
 
         public bool IsCleanSession { get; set; } = true;
@@ -44,5 +46,10 @@ namespace MQTTnet.Server.Internal
             status.PendingApplicationMessagesCount = ApplicationMessagesQueue.Count;
             status.Items = Items;
         }
+
+        public void OnDeleted()
+        {
+            Deleted?.Invoke(this, EventArgs.Empty);
+        }
     }
 }

Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs

@@ -265,6 +265,8 @@ namespace MQTTnet.Server.Internal
         public async Task DeleteSessionAsync(string clientId)
         {
             MqttClientConnection connection;
+            MqttClientSession session;
+            
             lock (_clientConnections)
             {
                 _clientConnections.TryGetValue(clientId, out connection);
@@ -272,6 +274,7 @@ namespace MQTTnet.Server.Internal
 
             lock (_clientSessions)
             {
+                _clientSessions.TryGetValue(clientId, out session);
                 _clientSessions.Remove(clientId);
             }
 
@@ -280,6 +283,8 @@ namespace MQTTnet.Server.Internal
                 await connection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
             }
 
+            session?.OnDeleted();
+            
             _logger.Verbose("Session for client '{0}' deleted.", clientId);
         }
 

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -29,7 +29,8 @@ namespace MQTTnet.Server.Internal
             _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
             _options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
             _eventDispatcher = eventDispatcher ?? throw new ArgumentNullException(nameof(eventDispatcher));
-            _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
+            _retainedMessagesManager = retainedMessagesManager ??
+                                       throw new ArgumentNullException(nameof(retainedMessagesManager));
         }
 
         public async Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket)
@@ -46,7 +47,7 @@ namespace MQTTnet.Server.Internal
                 var interceptorContext = await InterceptSubscribe(originalTopicFilter).ConfigureAwait(false);
                 var finalTopicFilter = interceptorContext.TopicFilter;
                 var acceptSubscription = interceptorContext.ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
-                
+
                 if (string.IsNullOrEmpty(finalTopicFilter.Topic) || !acceptSubscription)
                 {
                     // Return codes is for MQTT < 5 and Reason Code is for MQTT > 5.
@@ -69,56 +70,73 @@ namespace MQTTnet.Server.Internal
                     continue;
                 }
 
-                var subscription = CreateSubscription(
-                    finalTopicFilter, 
-                    subscribePacket.Properties?.SubscriptionIdentifier ?? 0,
-                    interceptorContext.ReasonCode);
+                if (interceptorContext.ProcessSubscription)
+                {
+                    var subscription = CreateSubscription(
+                        finalTopicFilter,
+                        subscribePacket.Properties?.SubscriptionIdentifier ?? 0,
+                        interceptorContext.ReasonCode);
 
-                await _eventDispatcher.SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter).ConfigureAwait(false);
+                    await _eventDispatcher
+                        .SafeNotifyClientSubscribedTopicAsync(_clientSession.ClientId, finalTopicFilter)
+                        .ConfigureAwait(false);
 
-                FilterRetainedApplicationMessages(retainedApplicationMessages, subscription, result);
+                    FilterRetainedApplicationMessages(retainedApplicationMessages, subscription, result);
+                }
             }
 
             return result;
         }
 
-        public async Task<List<MqttUnsubscribeReasonCode>> Unsubscribe(MqttUnsubscribePacket unsubscribePacket)
+        public async Task<UnsubscribeResult> Unsubscribe(MqttUnsubscribePacket unsubscribePacket)
         {
             if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
 
-            var reasonCodes = new List<MqttUnsubscribeReasonCode>();
+            var result = new UnsubscribeResult();
 
-            foreach (var topicFilter in unsubscribePacket.TopicFilters)
+            _subscriptionsLock.EnterWriteLock();
+            try
             {
-                var interceptorContext = await InterceptUnsubscribe(topicFilter).ConfigureAwait(false);
-                if (interceptorContext != null && !interceptorContext.AcceptUnsubscription)
+                foreach (var topicFilter in unsubscribePacket.TopicFilters)
                 {
-                    reasonCodes.Add(MqttUnsubscribeReasonCode.ImplementationSpecificError);
-                    continue;
-                }
+                    _subscriptions.TryGetValue(topicFilter, out var existingSubscription);
+                    
+                    var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription).ConfigureAwait(false);
+                    var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success;
 
-                _subscriptionsLock.EnterWriteLock();
-                try
-                {
-                    reasonCodes.Add(_subscriptions.Remove(topicFilter)
-                        ? MqttUnsubscribeReasonCode.Success
-                        : MqttUnsubscribeReasonCode.NoSubscriptionExisted);
-                }
-                finally
-                {
-                    _subscriptionsLock.ExitWriteLock();
+                    if (interceptorContext.CloseConnection)
+                    {
+                        result.CloseConnection = true;
+                    }
+                    
+                    if (!acceptUnsubscription)
+                    {
+                        result.ReasonCodes.Add(interceptorContext.ReasonCode);
+                        continue;
+                    }
+
+                    if (interceptorContext.ProcessUnsubscription)
+                    {
+                        _subscriptions.Remove(topicFilter);
+                    }
                 }
             }
+            finally
+            {
+                _subscriptionsLock.ExitWriteLock();
+            }
 
             foreach (var topicFilter in unsubscribePacket.TopicFilters)
             {
-                await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter)
+                    .ConfigureAwait(false);
             }
 
-            return reasonCodes;
+            return result;
         }
 
-        public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel, string senderClientId)
+        public CheckSubscriptionsResult CheckSubscriptions(string topic, MqttQualityOfServiceLevel qosLevel,
+            string senderClientId)
         {
             List<Subscription> subscriptions;
             _subscriptionsLock.EnterReadLock();
@@ -178,7 +196,8 @@ namespace MQTTnet.Server.Internal
             };
         }
 
-        Subscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier, MqttSubscribeReasonCode reasonCode)
+        Subscription CreateSubscription(MqttTopicFilter topicFilter, uint subscriptionIdentifier,
+            MqttSubscribeReasonCode reasonCode)
         {
             var subscription = new Subscription
             {
@@ -205,7 +224,7 @@ namespace MQTTnet.Server.Internal
             {
                 throw new InvalidOperationException();
             }
-            
+
             _subscriptionsLock.EnterWriteLock();
             try
             {
@@ -220,7 +239,8 @@ namespace MQTTnet.Server.Internal
             return subscription;
         }
 
-        static void FilterRetainedApplicationMessages(IList<MqttApplicationMessage> retainedApplicationMessages, Subscription subscription, SubscribeResult subscribeResult)
+        static void FilterRetainedApplicationMessages(IList<MqttApplicationMessage> retainedApplicationMessages,
+            Subscription subscription, SubscribeResult subscribeResult)
         {
             for (var i = retainedApplicationMessages.Count - 1; i >= 0; i--)
             {
@@ -236,13 +256,15 @@ namespace MQTTnet.Server.Internal
                     continue;
                 }
 
-                if (subscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly && !subscription.IsNewSubscription)
+                if (subscription.RetainHandling == MqttRetainHandling.SendAtSubscribeIfNewSubscriptionOnly &&
+                    !subscription.IsNewSubscription)
                 {
                     // This is a MQTT V5+ feature.
                     continue;
                 }
 
-                if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, subscription.Topic) != MqttTopicFilterCompareResult.IsMatch)
+                if (MqttTopicFilterComparer.Compare(retainedApplicationMessage.Topic, subscription.Topic) !=
+                    MqttTopicFilterCompareResult.IsMatch)
                 {
                     continue;
                 }
@@ -256,7 +278,7 @@ namespace MQTTnet.Server.Internal
 
                 if (subscription.Identifier > 0)
                 {
-                    queuedApplicationMessage.SubscriptionIdentifiers = new List<uint> { subscription.Identifier };
+                    queuedApplicationMessage.SubscriptionIdentifiers = new List<uint> {subscription.Identifier};
                 }
 
                 subscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage);
@@ -264,7 +286,7 @@ namespace MQTTnet.Server.Internal
                 retainedApplicationMessages[i] = null;
             }
         }
-        
+
         async Task<MqttSubscriptionInterceptorContext> InterceptSubscribe(MqttTopicFilter topicFilter)
         {
             var context = new MqttSubscriptionInterceptorContext
@@ -303,21 +325,15 @@ namespace MQTTnet.Server.Internal
                 var interceptor = _options.SubscriptionInterceptor;
                 if (interceptor != null)
                 {
-                    await interceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);    
+                    await interceptor.InterceptSubscriptionAsync(context).ConfigureAwait(false);
                 }
             }
-            
+
             return context;
         }
 
-        async Task<MqttUnsubscriptionInterceptorContext> InterceptUnsubscribe(string topicFilter)
+        async Task<MqttUnsubscriptionInterceptorContext> InterceptUnsubscribe(string topicFilter, Subscription subscription)
         {
-            var interceptor = _options.UnsubscriptionInterceptor;
-            if (interceptor == null)
-            {
-                return null;
-            }
-
             var context = new MqttUnsubscriptionInterceptorContext
             {
                 ClientId = _clientSession.ClientId,
@@ -325,12 +341,28 @@ namespace MQTTnet.Server.Internal
                 SessionItems = _clientSession.Items
             };
 
+            if (subscription == null)
+            {
+                context.DefaultReasonCode = MqttUnsubscribeReasonCode.NoSubscriptionExisted;
+            }
+            else
+            {
+                context.DefaultReasonCode = MqttUnsubscribeReasonCode.Success;
+            }
+
+            var interceptor = _options.UnsubscriptionInterceptor;
+            if (interceptor == null)
+            {
+                return context;
+            }
+
             await interceptor.InterceptUnsubscriptionAsync(context).ConfigureAwait(false);
 
             return context;
         }
 
-        static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel, HashSet<MqttQualityOfServiceLevel> subscribedQoSLevels)
+        static MqttQualityOfServiceLevel GetEffectiveQoS(MqttQualityOfServiceLevel qosLevel,
+            HashSet<MqttQualityOfServiceLevel> subscribedQoSLevels)
         {
             MqttQualityOfServiceLevel effectiveQoS;
             if (subscribedQoSLevels.Contains(qosLevel))

Source/MQTTnet/Server/Internal/UnsubscribeResult.cs

@@ -0,0 +1,12 @@
+using System.Collections.Generic;
+using MQTTnet.Protocol;
+
+namespace MQTTnet.Server.Internal
+{
+    public sealed class UnsubscribeResult
+    {
+        public List<MqttUnsubscribeReasonCode> ReasonCodes { get; } = new List<MqttUnsubscribeReasonCode>(128);
+        
+        public bool CloseConnection { get; set; }
+    }
+}

Source/MQTTnet/Server/MqttSubscriptionInterceptorContext.cs

@@ -25,12 +25,13 @@ namespace MQTTnet.Server
 
         // 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;
+            get => ReasonCode == MqttSubscribeReasonCode.GrantedQoS0 && ReasonCode <= MqttSubscribeReasonCode.GrantedQoS2;
             set
             {
                 if (value)
@@ -46,6 +47,13 @@ namespace MQTTnet.Server
             }
         }
 
+        /// <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_.
@@ -56,6 +64,9 @@ namespace MQTTnet.Server
         // MQTT < 5 only!
         internal MqttSubscribeReturnCode ReturnCode { get; set; }
 
+        /// <summary>
+        /// Gets or sets whether the broker should close the client connection.
+        /// </summary>
         public bool CloseConnection { get; set; }
     }
 }

Source/MQTTnet/Server/MqttUnsubscriptionInterceptorContext.cs

@@ -1,4 +1,6 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
+using MQTTnet.Protocol;
 
 namespace MQTTnet.Server
 {
@@ -22,8 +24,32 @@ namespace MQTTnet.Server
         /// </summary>
         public IDictionary<object, object> SessionItems { get; internal set; }
 
-        public bool AcceptUnsubscription { get; set; } = true;
+        [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.
+        /// </summary>
+        public MqttUnsubscribeReasonCode ReasonCode { get; set; }
+        
+        /// <summary>
+        /// Gets or sets whether the broker should remove 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 ProcessUnsubscription { get; set; } = true;
+        
+        /// <summary>
+        /// Gets or sets whether the broker should close the client connection.
+        /// </summary>
         public bool CloseConnection { get; set; }
     }
 }

Source/MQTTnet/Server/Status/MqttSessionStatus.cs

@@ -14,8 +14,15 @@ namespace MQTTnet.Server.Status
         {
             _session = session ?? throw new ArgumentNullException(nameof(session));
             _sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
+
+            session.Deleted += (_, __) => Deleted?.Invoke(this, EventArgs.Empty);
         }
 
+        /// <summary>
+        /// Fired when this session is being deleted.
+        /// </summary>
+        public event EventHandler Deleted;
+        
         /// <summary>
         /// Gets or sets the client identifier.
         /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.

Tests/MQTTnet.Core.Tests/Internal/AsyncLock_Tests.cs

@@ -4,7 +4,7 @@ using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Internal;
 
-namespace MQTTnet.Tests
+namespace MQTTnet.Tests.Internal
 {
     [TestClass]
     public class AsyncLock_Tests

Tests/MQTTnet.Core.Tests/Internal/AsyncQueue_Tests.cs

@@ -4,7 +4,7 @@ using System.Threading.Tasks;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Internal;
 
-namespace MQTTnet.Tests
+namespace MQTTnet.Tests.Internal
 {
     [TestClass]
     public class AsyncQueue_Tests

Tests/MQTTnet.Core.Tests/Internal/MqttPacketBus_Tests.cs

@@ -0,0 +1,110 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using MQTTnet.Internal;
+using MQTTnet.Packets;
+
+namespace MQTTnet.Tests.Internal
+{
+    [TestClass]
+    public sealed class MqttPacketBus_Tests
+    {
+        [TestMethod]
+        [ExpectedException(typeof(OperationCanceledException))]
+        public async Task Wait_With_Empty_Bus()
+        {
+            var bus = new MqttPacketBus();
+
+            using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)))
+            {
+                await bus.DequeueAsync(timeout.Token);
+            }
+        }
+        
+        [TestMethod]
+        public void Alternate_Priorities()
+        {
+            var bus = new MqttPacketBus();
+            
+            bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            
+            bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
+            bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
+            bus.Enqueue(new MqttPacketBusItem(new MqttSubAckPacket()), MqttPacketBusPartition.Control);
+            
+            bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
+            bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
+            bus.Enqueue(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health);
+
+            Assert.AreEqual(9, bus.PacketsCount);
+            
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket));
+            Assert.IsInstanceOfType(bus.DequeueAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket));
+            
+            Assert.AreEqual(0, bus.PacketsCount);
+        }
+        
+        [TestMethod]
+        public void Export_Packets_Without_Dequeue()
+        {
+            var bus = new MqttPacketBus();
+
+            bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            bus.Enqueue(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data);
+            
+            Assert.AreEqual(3, bus.PacketsCount);
+
+            var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control);
+            Assert.AreEqual(0, exportedPackets.Count);
+            
+            exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Health);
+            Assert.AreEqual(0, exportedPackets.Count);
+            
+            exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data);
+            Assert.AreEqual(3, exportedPackets.Count);
+            
+            Assert.AreEqual(3, bus.PacketsCount);
+        }
+        
+        [TestMethod]
+        public void Await_Single_Packet()
+        {
+            var bus = new MqttPacketBus();
+
+            var delivered = false;
+            
+            bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data);
+            bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data);
+            bus.Enqueue(new MqttPublishPacket(), MqttPacketBusPartition.Data).Delivered += (_, __) =>
+            {
+                delivered = true;
+            };
+            
+            Assert.IsFalse(delivered);
+
+            bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered();
+            
+            Assert.IsFalse(delivered);
+            
+            bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered();
+            
+            Assert.IsFalse(delivered);
+            
+            bus.DequeueAsync(CancellationToken.None).Result.MarkAsDelivered();
+            
+            // The third packet has the event attached.
+            Assert.IsTrue(delivered);
+        }
+    }
+}