diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index da84f415..47ac9038 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/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 HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) + async Task 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) diff --git a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs index bd80dc60..8e4f2f54 100644 --- a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs +++ b/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; + /// /// 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 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; + } + } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs index 36daed4c..c28fdf9e 100644 --- a/Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs +++ b/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)