Merge pull request #40 from JanEggers/develop

Develop

MQTTnet.Core/Client/MqttClient.cs

@@ -249,21 +249,25 @@ namespace MQTTnet.Core.Client
                 if (mqttPacket is MqttPingReqPacket)
                 {
                     await SendAsync(new MqttPingRespPacket());
+                    return;
                 }
 
                 if (mqttPacket is MqttDisconnectPacket)
                 {
                     await DisconnectAsync();
+                    return;
                 }
 
                 if (mqttPacket is MqttPublishPacket publishPacket)
                 {
                     await ProcessReceivedPublishPacket(publishPacket);
+                    return;
                 }
 
                 if (mqttPacket is MqttPubRelPacket pubRelPacket)
                 {
                     await ProcessReceivedPubRelPacket(pubRelPacket);
+                    return;
                 }
 
                 _packetDispatcher.Dispatch(mqttPacket);
@@ -336,8 +340,9 @@ namespace MQTTnet.Core.Client
 
         private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
         {
+            var wait = _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout);
             await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, requestPacket).ConfigureAwait(false);
-            return (TResponsePacket)await _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout).ConfigureAwait(false);
+            return (TResponsePacket)await wait.ConfigureAwait(false);
         }
 
         private ushort GetNewPacketIdentifier()

MQTTnet.Core/Server/MqttClientMessageQueue.cs

@@ -54,6 +54,12 @@ namespace MQTTnet.Core.Server
             var consumable = _pendingPublishPackets.GetConsumingEnumerable();
             while (!cancellationToken.IsCancellationRequested)
             {
+                if (_pendingPublishPackets.Count == 0)
+                {
+                    await Task.Delay(TimeSpan.FromMilliseconds(5)).ConfigureAwait(false);
+                    continue;
+                }
+
                 var packets = consumable.Take(_pendingPublishPackets.Count).ToList();
                 try
                 {