ClientMessageQueueInterceptor continues with subscription processing. (#1685)

* ClientMessageQueueInterceptor continues with subscription processing.

* Update readme

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
This commit is contained in:
RossHNPC
2023-04-06 17:55:33 +02:00
committed by GitHub
co-authored by Christian
parent 36c6d8974e
commit 2933b76237
3 changed files with 42 additions and 5 deletions
+1
View File
@@ -13,4 +13,5 @@
* [Server] Added a new API method which allows reading a single retained message without the need to processing the entire set of retained messages (#1659). * [Server] Added a new API method which allows reading a single retained message without the need to processing the entire set of retained messages (#1659).
* [Server] Added a new event (InterceptingClientEnqueueAsync) which allows intercepting when an application message is enqueued for a client (#1648). * [Server] Added a new event (InterceptingClientEnqueueAsync) which allows intercepting when an application message is enqueued for a client (#1648).
* [Server] Fixed race condition when handling connections which leads to stopped message transfers (#1677, thanks to @RazvanEmilR). * [Server] Fixed race condition when handling connections which leads to stopped message transfers (#1677, thanks to @RazvanEmilR).
* [Server] Fixed not delivered packets to other sessions if the _InterceptingClientEnqueueEvent_ is used and accepting the packet is denied (#1685, thanks to @RossHNPC).
* [Server] Added missing user properties to subscribe and unsubscribe interceptor (#1702). * [Server] Added missing user properties to subscribe and unsubscribe interceptor (#1702).
@@ -105,5 +105,42 @@ namespace MQTTnet.Tests.Server
receivedMessages.AssertReceivedCountEquals(1); receivedMessages.AssertReceivedCountEquals(1);
} }
} }
[TestMethod]
public async Task Intercept_Client_Enqueue_Multiple_Clients_Subscribed_Messages_Are_Filtered()
{
using (var testEnvironment = CreateTestEnvironment())
{
var server = await testEnvironment.StartServer();
var sender = await testEnvironment.ConnectClient();
var receiverOne = await testEnvironment.ConnectClient(o => o.WithClientId("One"));
await receiverOne.SubscribeAsync("A");
var receivedMessagesOne = testEnvironment.CreateApplicationMessageHandler(receiverOne);
var receiverTwo = await testEnvironment.ConnectClient(o => o.WithClientId("Two"));
await receiverTwo.SubscribeAsync("A");
var receivedMessagesTwo = testEnvironment.CreateApplicationMessageHandler(receiverTwo);
var receiverThree = await testEnvironment.ConnectClient(o => o.WithClientId("Three"));
await receiverThree.SubscribeAsync("A");
var receivedMessagesThree = testEnvironment.CreateApplicationMessageHandler(receiverThree);
server.InterceptingClientEnqueueAsync += e =>
{
if (e.ReceiverClientId.Contains("Two")) e.AcceptEnqueue = false;
return CompletedTask.Instance;
};
await sender.PublishStringAsync("A", "Payload", MqttQualityOfServiceLevel.AtLeastOnce);
await LongTestDelay();
receivedMessagesOne.AssertReceivedCountEquals(1);
receivedMessagesTwo.AssertReceivedCountEquals(0);
receivedMessagesThree.AssertReceivedCountEquals(1);
}
}
} }
} }
@@ -207,9 +207,8 @@ namespace MQTTnet.Server
if (!eventArgs.AcceptEnqueue) if (!eventArgs.AcceptEnqueue)
{ {
// There will be no reason string and use properties because in this case the clients will // Continue checking the other subscriptions
// not receive a packet at all. continue;
return new DispatchApplicationMessageResult(reasonCode, eventArgs.CloseSenderConnection, null, null);
} }
} }