YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-11-05 22:16:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add flag for indicating if application message processing has failed.

Browse Source
This commit is contained in:
Christian Kratky
2019-03-10 09:51:49 +01:00
parent 6e64d653c5
commit 9070087fad
2 changed files with 22 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Client/MqttClient.cs
+20 -17
View File
@@ -510,25 +510,27 @@ namespace MQTTnet.Client
} }
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{ {
await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
await SendAsync(new MqttPubAckPacket
{ {
PacketIdentifier = publishPacket.PacketIdentifier, await SendAsync(new MqttPubAckPacket
ReasonCode = MqttPubAckReasonCode.Success {
}, cancellationToken).ConfigureAwait(false); PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
}
} }
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{ {
await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false); if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
var pubRecPacket = new MqttPubRecPacket
{ {
PacketIdentifier = publishPacket.PacketIdentifier, var pubRecPacket = new MqttPubRecPacket
ReasonCode = MqttPubRecReasonCode.Success {
}; PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRecReasonCode.Success
};
await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false); await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
}
} }
else else
{ {
@@ -572,18 +574,19 @@ namespace MQTTnet.Client
return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket); return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket);
} }
private Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket) private async Task<bool> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
{ {
var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket); var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
var handler = ApplicationMessageReceivedHandler; var handler = ApplicationMessageReceivedHandler;
if (handler != null) if (handler != null)
{ {
return handler.HandleApplicationMessageReceivedAsync( var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage);
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) private static async Task WaitForTaskAsync(Task task, Task sender)
Source/MQTTnet/MqttApplicationMessageReceivedEventArgs.cs
+2
View File
@@ -13,5 +13,7 @@ namespace MQTTnet
public string ClientId { get; } public string ClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; } public MqttApplicationMessage ApplicationMessage { get; }
public bool ProcessingFailed { get; set; }
} }
} }