diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 271e6b93..7647fc7f 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -254,7 +254,15 @@ namespace MQTTnet.Extensions.ManagedClient { while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected) { - var message = _messageQueue.Dequeue(); + //Peek at the message without dequeueing in order to prevent the + //possibility of the queue growing beyond the configured cap. + //Previously, messages could be re-enqueued if there was an + //exception, and this re-enqueueing did not honor the cap. + //Furthermore, because re-enqueueing would shuffle the order + //of the messages, the DropOldestQueuedMessage strategy would + //be unable to know which message is actually the oldest and would + //instead drop the first item in the queue. + var message = _messageQueue.PeekAndWait(); if (message == null) { continue; @@ -284,6 +292,16 @@ namespace MQTTnet.Extensions.ManagedClient try { _mqttClient.PublishAsync(message.ApplicationMessage).GetAwaiter().GetResult(); + lock (_messageQueue) //lock to avoid conflict with this.PublishAsync + { + //While publishing this message, this.PublishAsync could have booted this + //message off the queue to make room for another (when using a cap + //with the DropOldestQueuedMessage strategy). If the first item + //in the queue is equal to this message, then it's safe to remove + //it from the queue. If not, that means this.PublishAsync has already + //removed it, in which case we don't want to do anything. + _messageQueue.RemoveFirstIfEqual(message, IdsAreEqual); + } _storageManager?.RemoveAsync(message).GetAwaiter().GetResult(); } catch (MqttCommunicationException exception) @@ -292,9 +310,19 @@ namespace MQTTnet.Extensions.ManagedClient _logger.Warning(exception, $"Publishing application ({message.Id}) message failed."); - if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + if (message.ApplicationMessage.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) { - _messageQueue.Enqueue(message); + //If QoS 0, we don't want this message to stay on the queue. + //If QoS 1 or 2, it's possible that, when using a cap, this message + //has been booted off the queue by this.PublishAsync, in which case this + //thread will not continue to try to publish it. While this does + //contradict the expected behavior of QoS 1 and 2, that's also true + //for the usage of a message queue cap, so it's still consistent + //with prior behavior in that way. + lock (_messageQueue) //lock to avoid conflict with this.PublishAsync + { + _messageQueue.RemoveFirstIfEqual(message, IdsAreEqual); + } } } catch (Exception exception) @@ -307,6 +335,11 @@ namespace MQTTnet.Extensions.ManagedClient ApplicationMessageProcessed?.Invoke(this, new ApplicationMessageProcessedEventArgs(message, transmitException)); } } + + private static bool IdsAreEqual(ManagedMqttApplicationMessage message1, ManagedMqttApplicationMessage message2) + { + return message1.Id.Equals(message2.Id); + } private async Task SynchronizeSubscriptionsAsync() { diff --git a/Source/MQTTnet/Internal/BlockingQueue.cs b/Source/MQTTnet/Internal/BlockingQueue.cs index d6c28f46..2812349a 100644 --- a/Source/MQTTnet/Internal/BlockingQueue.cs +++ b/Source/MQTTnet/Internal/BlockingQueue.cs @@ -55,6 +55,38 @@ namespace MQTTnet.Internal _gate.WaitOne(); } } + + public TItem PeekAndWait() + { + while (true) + { + lock (_syncRoot) + { + if (_items.Count > 0) + { + return _items.First.Value; + } + + if (_items.Count == 0) + { + _gate.Reset(); + } + } + + _gate.WaitOne(); + } + } + + public void RemoveFirstIfEqual(TItem item, Func areEqual) + { + lock (_syncRoot) + { + if (_items.Count > 0 && areEqual(_items.First.Value, item)) + { + _items.RemoveFirst(); + } + } + } public TItem RemoveFirst() {