diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 45b99c88..a09f6b53 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -510,25 +510,27 @@ namespace MQTTnet.Client } else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) { - await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); - - await SendAsync(new MqttPubAckPacket + if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false)) { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubAckReasonCode.Success - }, cancellationToken).ConfigureAwait(false); + await SendAsync(new MqttPubAckPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = MqttPubAckReasonCode.Success + }, cancellationToken).ConfigureAwait(false); + } } else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) { - await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); - - var pubRecPacket = new MqttPubRecPacket + if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false)) { - PacketIdentifier = publishPacket.PacketIdentifier, - ReasonCode = MqttPubRecReasonCode.Success - }; + var pubRecPacket = new MqttPubRecPacket + { + PacketIdentifier = publishPacket.PacketIdentifier, + ReasonCode = MqttPubRecReasonCode.Success + }; - await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false); + await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false); + } } else { @@ -572,18 +574,19 @@ namespace MQTTnet.Client return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket); } - private Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) + private async Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) { var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket); var handler = ApplicationMessageReceivedHandler; if (handler != null) { - return handler.HandleApplicationMessageReceivedAsync( - new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage)); + var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage); + await handler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false); + return !eventArgs.ProcessingFailed; } - return Task.FromResult(0); + return true; } private static async Task WaitForTaskAsync(Task task, Task sender) diff --git a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs index e81c69a9..6fbaa933 100644 --- a/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs +++ b/Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs @@ -13,5 +13,7 @@ namespace MQTTnet public string ClientId { get; } public MqttApplicationMessage ApplicationMessage { get; } + + public bool ProcessingFailed { get; set; } } }