Improve Unit Tests.

Source/MQTTnet/PacketDispatcher/MqttPacketAwaitable.cs

@@ -3,12 +3,13 @@ using MQTTnet.Packets;
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using MQTTnet.Internal;
 
 namespace MQTTnet.PacketDispatcher
 {
     public sealed class MqttPacketAwaitable<TPacket> : IMqttPacketAwaitable where TPacket : MqttBasePacket
     {
-        readonly TaskCompletionSource<MqttBasePacket> _taskCompletionSource;
+        readonly CrossPlatformPromise<MqttBasePacket> _promise = new CrossPlatformPromise<MqttBasePacket>();
         readonly MqttPacketDispatcher _owningPacketDispatcher;
 
         public MqttPacketAwaitable(ushort packetIdentifier, MqttPacketDispatcher owningPacketDispatcher)
@@ -20,11 +21,6 @@ namespace MQTTnet.PacketDispatcher
             };
             
             _owningPacketDispatcher = owningPacketDispatcher ?? throw new ArgumentNullException(nameof(owningPacketDispatcher));
-#if NET452
-            _taskCompletionSource = new TaskCompletionSource<MqttBasePacket>();
-#else
-            _taskCompletionSource = new TaskCompletionSource<MqttBasePacket>(TaskCreationOptions.RunContinuationsAsynchronously);
-#endif
         }
         
         public MqttPacketAwaitableFilter Filter { get; }
@@ -35,7 +31,7 @@ namespace MQTTnet.PacketDispatcher
             {
                 using (timeoutToken.Token.Register(() => Fail(new MqttCommunicationTimedOutException())))
                 {
-                    var packet = await _taskCompletionSource.Task.ConfigureAwait(false);
+                    var packet = await _promise.Task.ConfigureAwait(false);
                     return (TPacket)packet;
                 }
             }
@@ -45,45 +41,19 @@ namespace MQTTnet.PacketDispatcher
         {
             if (packet == null) throw new ArgumentNullException(nameof(packet));
 
-#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
-            Task.Run(() => _taskCompletionSource.TrySetResult(packet));
-#else
-            _taskCompletionSource.TrySetResult(packet);
-#endif
+            _promise.TrySetResult(packet);
         }
 
         public void Fail(Exception exception)
         {
             if (exception == null) throw new ArgumentNullException(nameof(exception));
-#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
-            Task.Run(() => _taskCompletionSource.TrySetException(exception));
-#else
-            _taskCompletionSource.TrySetException(exception);
-#endif
+            
+            _promise.TrySetException(exception);
         }
 
         public void Cancel()
         {
-#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
-            Task.Run(() => _taskCompletionSource.TrySetCanceled());
-#else
-            _taskCompletionSource.TrySetCanceled();
-#endif
+            _promise.TrySetCanceled();
         }
 
         public void Dispose()

Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs

@@ -62,9 +62,11 @@ namespace MQTTnet.Server.Internal
 
                 if (interceptorContext.CloseConnection)
                 {
+                    // When any of the interceptor calls leads to a connection close the connection
+                    // must be closed. So do not revert to false!
                     result.CloseConnection = true;
                 }
-
+                
                 if (!acceptSubscription || string.IsNullOrEmpty(finalTopicFilter.Topic))
                 {
                     continue;
@@ -104,14 +106,17 @@ namespace MQTTnet.Server.Internal
                     var interceptorContext = await InterceptUnsubscribe(topicFilter, existingSubscription).ConfigureAwait(false);
                     var acceptUnsubscription = interceptorContext.ReasonCode == MqttUnsubscribeReasonCode.Success;
 
+                    result.ReasonCodes.Add(interceptorContext.ReasonCode);
+
                     if (interceptorContext.CloseConnection)
                     {
+                        // When any of the interceptor calls leads to a connection close the connection
+                        // must be closed. So do not revert to false!
                         result.CloseConnection = true;
                     }
-                    
+           
                     if (!acceptUnsubscription)
                     {
-                        result.ReasonCodes.Add(interceptorContext.ReasonCode);
                         continue;
                     }
 
@@ -128,8 +133,7 @@ namespace MQTTnet.Server.Internal
 
             foreach (var topicFilter in unsubscribePacket.TopicFilters)
             {
-                await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter)
-                    .ConfigureAwait(false);
+                await _eventDispatcher.SafeNotifyClientUnsubscribedTopicAsync(_clientSession.ClientId, topicFilter).ConfigureAwait(false);
             }
 
             return result;

Source/MQTTnet/Server/MqttServerOptionsBuilder.cs

@@ -249,12 +249,12 @@ namespace MQTTnet.Server
             return this;
         }
 
-        public MqttServerOptionsBuilder WithPersistentSessions()
+        public MqttServerOptionsBuilder WithPersistentSessions(bool value = true)
         {
-            _options.EnablePersistentSessions = true;
+            _options.EnablePersistentSessions = value;
             return this;
         }
-
+        
         /// <summary>
         /// Gets or sets the client ID which is used when publishing messages from the server directly.
         /// </summary>

Source/MQTTnet/Server/Status/IMqttSessionStatus.cs

@@ -1,4 +1,5 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace MQTTnet.Server.Status
@@ -6,7 +7,12 @@ namespace MQTTnet.Server.Status
     public interface IMqttSessionStatus
     {
         /// <summary>
-        /// Gets the client identifier.
+        /// Fired when this session is being deleted.
+        /// </summary>
+        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.
         /// </summary>
         string ClientId { get; }
@@ -16,10 +22,27 @@ namespace MQTTnet.Server.Status
         /// </summary>
         long PendingApplicationMessagesCount { get; }
 
+        /// <summary>
+        /// This items can be used by the library user in order to store custom information.
+        /// </summary>
         IDictionary<object, object> Items { get; }
 
-        Task ClearPendingApplicationMessagesAsync();
+        /// <summary>
+        /// Enqueues a new application message.
+        /// </summary>
+        /// <param name="applicationMessage">The application message.</param>
+        /// <returns>A task.</returns>
+        Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage);
+        
+        /// <summary>
+        /// Clears the entire queue with pending application messages.
+        /// </summary>
+        /// <returns>A task.</returns>
+        Task ClearApplicationMessagesQueueAsync();
 
+        [Obsolete("Please use _ClearApplicationMessagesQueueAsync_ instead. This will be removed soon.")]
+        Task ClearPendingApplicationMessagesAsync();
+        
         Task DeleteAsync();
     }
 }

Source/MQTTnet/Server/Status/MqttSessionStatus.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using MQTTnet.Implementations;
 using MQTTnet.Server.Internal;
 
 namespace MQTTnet.Server.Status
@@ -17,36 +18,47 @@ namespace MQTTnet.Server.Status
 
             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.
-        /// </summary>
         public string ClientId { get; set; }
 
         public long PendingApplicationMessagesCount { get; set; }
 
         public DateTime CreatedTimestamp { get; set; }
-
-        /// <summary>
-        /// This items can be used by the library user in order to store custom information.
-        /// </summary>
+        
         public IDictionary<object, object> Items { get; set; }
         
-        public Task DeleteAsync()
+        public Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage)
         {
-            return _sessionsManager.DeleteSessionAsync(ClientId);
+            if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
+            
+            _session.ApplicationMessagesQueue.Enqueue(new MqttQueuedApplicationMessage
+            {
+                ApplicationMessage = applicationMessage,
+                IsDuplicate = false,
+                IsRetainedMessage = false,
+                SubscriptionQualityOfServiceLevel = applicationMessage.QualityOfServiceLevel,
+                SenderClientId = null
+            });
+
+            return PlatformAbstractionLayer.CompletedTask;
         }
 
-        public Task ClearPendingApplicationMessagesAsync()
+        public Task ClearApplicationMessagesQueueAsync()
         {
             _session.ApplicationMessagesQueue.Clear();
             return Task.FromResult(0);
         }
+
+        public Task DeleteAsync()
+        {
+            return _sessionsManager.DeleteSessionAsync(ClientId);
+        }
+        
+        public Task ClearPendingApplicationMessagesAsync()
+        {
+            return ClearApplicationMessagesQueueAsync();
+        }
     }
 }