Fixes chkr1011's refactoing and pass all tests

Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs

@@ -136,9 +136,9 @@ namespace MQTTnet.Server
             return Task.FromResult((IList<IMqttClientSessionStatus>)result);
         }
 
-        public void StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
+        public Task StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage)
         {
-            Task.Run(() => DispatchApplicationMessageAsync(senderClientSession, applicationMessage));
+            return DispatchApplicationMessageAsync(senderClientSession, applicationMessage);
         }
 
         public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
@@ -237,6 +237,7 @@ namespace MQTTnet.Server
         {
             try
             {
+                _logger.Verbose($"DispatchApplicationMessageAsync");
                 var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage);
                 if (interceptorContext.CloseConnection)
                 {
@@ -257,6 +258,7 @@ namespace MQTTnet.Server
 
                 foreach (var clientSession in _sessions.Values)
                 {
+                    _logger.Verbose($"DispatchApplicationMessageAsync for {clientSession.ClientId} ");
                     clientSession.EnqueueApplicationMessage(applicationMessage);
                 }
             }

Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs

@@ -82,7 +82,7 @@ namespace MQTTnet.Server
         {
             var saveIsRequired = false;
 
-            if (applicationMessage.Payload?.Length > 0)
+            if (applicationMessage.Payload?.Length == 0)
             {
                 saveIsRequired = _messages.TryRemove(applicationMessage.Topic, out _);
                 _logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);