reimplemented delayed acknowledgement using a property to switch off automatic acknowledgement and a new method Acknowledge() in the event handler

Source/MQTTnet/Client/MqttClient.cs

@@ -641,32 +641,13 @@ namespace MQTTnet.Client
                     }
 
                     var publishPacket = publishPacketDequeueResult.Item;
-                    var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket);
+                    var eventArgs = await HandleReceivedApplicationMessageAsync(publishPacket, cancellationToken);
 
-                    if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
+                    if (eventArgs.AutoAcknowledge) 
                     {
-                        // no response required
-                    }
-                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
-                    {
-                        if (!eventArgs.ProcessingFailed)
-                        {
-                            var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(publishPacket, eventArgs.ReasonCode);
-                            await SendResponseForReceivedPublishPacket(eventArgs, pubAckPacket, cancellationToken).ConfigureAwait(false);
-                        }
-                    }
-                    else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
-                    {
-                        if (!eventArgs.ProcessingFailed)
-                        {
-                            var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(publishPacket, eventArgs.ReasonCode);
-                            await SendResponseForReceivedPublishPacket(eventArgs, pubRecPacket, cancellationToken).ConfigureAwait(false);
-                        }
-                    }
-                    else
-                    {
-                        throw new MqttProtocolViolationException("Received a not supported QoS level.");
+                        await eventArgs.Acknowledge();
                     }
+                    
                 }
                 catch (OperationCanceledException)
                 {
@@ -678,29 +659,34 @@ namespace MQTTnet.Client
             }
         }
 
-        async Task SendResponseForReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs, MqttBasePacket resultPacket, CancellationToken cancellationToken)
+        internal Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs eventArgs)
         {
-            if (eventArgs.PendingTask == null)
+            if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
             {
-                await SendAsync(resultPacket, cancellationToken).ConfigureAwait(false);
+                // no response required
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubAckPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubAckPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
+                    return SendAsync(pubAckPacket, eventArgs.CancellationToken);
+                }
+            }
+            else if (eventArgs.PublishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
+            {
+                if (!eventArgs.ProcessingFailed)
+                {
+                    var pubRecPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePubRecPacket(eventArgs.PublishPacket, eventArgs.ReasonCode);
+                    return SendAsync(pubRecPacket, eventArgs.CancellationToken);
+                }
             }
             else
             {
-                _ = eventArgs.PendingTask.ContinueWith(async x =>
-                {
-                    try
-                    {
-                        if (x.Result) await SendAsync(resultPacket, cancellationToken).ConfigureAwait(false);
-                    }
-                    catch (OperationCanceledException)
-                    {
-                    }
-                    catch (Exception exception)
-                    {
-                        _logger.Error(exception, "Error while handling application message.");
-                    }
-                });
+                throw new MqttProtocolViolationException("Received a not supported QoS level.");
             }
+
+            return PlatformAbstractionLayer.CompletedTask;
         }
 
         Task ProcessReceivedPubRecPacket(MqttPubRecPacket pubRecPacket, CancellationToken cancellationToken)
@@ -775,10 +761,10 @@ namespace MQTTnet.Client
             return _adapter.PacketFormatterAdapter.DataConverter.CreateClientPublishResult(pubRecPacket, pubCompPacket);
         }
 
-        async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
+        async Task<MqttApplicationMessageReceivedEventArgs> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken)
         {
             var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
-            var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage);
+            var eventArgs = new MqttApplicationMessageReceivedEventArgs(this, publishPacket, cancellationToken, Options.ClientId, applicationMessage);
 
             var handler = ApplicationMessageReceivedHandler;
             if (handler != null)

Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs

@@ -1,16 +1,41 @@
-using System;
+using MQTTnet.Client;
+using MQTTnet.Implementations;
+using MQTTnet.Packets;
+using System;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace MQTTnet
 {
     public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
     {
-        public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
+        public MqttApplicationMessageReceivedEventArgs(MqttClient client, MqttPublishPacket publishPacket, CancellationToken cancellationToken, string clientId, MqttApplicationMessage applicationMessage)
         {
+            Client = client ?? throw new ArgumentNullException(nameof(client));
+            PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket));
+            CancellationToken = cancellationToken;
+            acknowledged = 0;
             ClientId = clientId;
             ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            AutoAcknowledge = true;
         }
 
+        public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
+        {
+            acknowledged = 1;
+            ClientId = clientId;
+            ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
+            AutoAcknowledge = true;
+        }
+
+        internal MqttClient Client { get; }
+
+        internal CancellationToken CancellationToken { get; }
+
+        internal MqttPublishPacket PublishPacket { get; }
+
+        int acknowledged;
+
         /// <summary>
         /// Gets the client identifier.
         /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
@@ -31,6 +56,18 @@ namespace MQTTnet
         
         public object Tag { get; set; }
 
-        public Task<bool> PendingTask { get; set; }
+        public bool AutoAcknowledge { get; set; }
+
+        public Task Acknowledge() 
+        { 
+            if (Interlocked.CompareExchange(ref acknowledged, 1, 0) == 0)
+            {
+                return Client.AcknowledgeReceivedPublishPacket(this);
+            }
+            else
+            {
+                return PlatformAbstractionLayer.CompletedTask;
+            }
+        }
     }
 }

Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs

@@ -465,7 +465,8 @@ namespace MQTTnet.Tests
                 Task Handler1(MqttApplicationMessageReceivedEventArgs eventArgs)
                 {
                     var value = int.Parse(eventArgs.ApplicationMessage.ConvertPayloadToString());
-                    eventArgs.PendingTask = Task.Delay(value).ContinueWith(x => true);
+                    eventArgs.AutoAcknowledge = false;
+                    Task.Delay(value).ContinueWith(x => eventArgs.Acknowledge());
 
                     System.Diagnostics.Debug.WriteLine($"received {value}");
                     lock (receivedValues)