YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-16 17:23:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix wrong QoS when using MqttClientMessageQueueInterceptor #978

Browse Source
This commit is contained in:
Marc Sallin
2020-09-04 09:53:34 +02:00
committed by Sallin Marc, I212
parent fd789417e7
commit 7f5504b01b
2 changed files with 34 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/MqttClientConnection.cs
+8 -7
View File
@@ -397,17 +397,13 @@ namespace MQTTnet.Server
// Set the retain flag to true according to [MQTT-3.3.1-8] and [MQTT-3.3.1-9].
publishPacket.Retain = queuedApplicationMessage.IsRetainedMessage;
if (publishPacket.QualityOfServiceLevel > 0)
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
}
if (_serverOptions.ClientMessageQueueInterceptor != null)
{
var context = new MqttClientMessageQueueInterceptorContext(
queuedApplicationMessage.SenderClientId,
ClientId,
queuedApplicationMessage.ApplicationMessage);
queuedApplicationMessage.ApplicationMessage,
queuedApplicationMessage.SubscriptionQualityOfServiceLevel);
if (_serverOptions.ClientMessageQueueInterceptor != null)
{
@@ -421,7 +417,12 @@ namespace MQTTnet.Server
publishPacket.Topic = context.ApplicationMessage.Topic;
publishPacket.Payload = context.ApplicationMessage.Payload;
publishPacket.QualityOfServiceLevel = context.ApplicationMessage.QualityOfServiceLevel;
publishPacket.QualityOfServiceLevel = context.SubscriptionQualityOfServiceLevel;
}
if (publishPacket.QualityOfServiceLevel > 0)
{
publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNextPacketIdentifier();
}
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
Tests/MQTTnet.Core.Tests/Server_Tests.cs
+26
View File
@@ -1452,5 +1452,31 @@ namespace MQTTnet.Tests
}
}
[TestMethod]
public async Task Intercept_ClientMessageQueue_Different_QoS_Of_Subscription_And_Message()
{
const string topic = "a";
using (var testEnvironment = new TestEnvironment(TestContext))
{
await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder()
.WithClientMessageQueueInterceptor(c => { })); // Interceptor does nothing but has to be present.
bool receivedMessage = false;
var client = await testEnvironment.ConnectClientAsync();
client.UseApplicationMessageReceivedHandler(c =>
{
receivedMessage = true;
});
await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce);
await client.PublishAsync(new MqttApplicationMessage{ Topic = topic, QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
await Task.Delay(500);
Assert.IsTrue(receivedMessage);
}
}
}
}