This is an automated email from the ASF dual-hosted git repository.
tabish121 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
The following commit(s) were added to refs/heads/main by this push:
new 8fb83232 PROTON-2951 Fix issue in next receiver processing return order
8fb83232 is described below
commit 8fb832324b35068c4d52b5cb54c94f5e5ab999c3
Author: Timothy Bish <[email protected]>
AuthorDate: Mon Aug 10 12:37:03 2026 -0400
PROTON-2951 Fix issue in next receiver processing return order
Ensure correct receiver is returned when using round robin policy for
selecting the next available receiver
---
.../client/impl/ClientNextReceiverSelector.java | 5 +-
.../qpid/protonj2/client/impl/SessionTest.java | 91 ++++++++++++++++++++++
2 files changed, 94 insertions(+), 2 deletions(-)
diff --git
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java
index 8fc27a14..c704778a 100644
---
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java
+++
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientNextReceiverSelector.java
@@ -146,9 +146,10 @@ final class ClientNextReceiverSelector {
for (org.apache.qpid.protonj2.engine.Receiver protonReceover :
session.getProtonSession().receivers()) {
if (protonReceover.getLinkedResource() instanceof
ClientReceiver) {
if (foundLast) {
- ClientReceiver candidate =
protonReceover.getLinkedResource();
+ final ClientReceiver candidate =
protonReceover.getLinkedResource();
+
if (candidate.queuedDeliveries() > 0) {
- result = candidate;
+ return candidate;
}
} else {
foundLast = protonReceover.getLinkedResource() ==
lastReceiver;
diff --git
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SessionTest.java
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SessionTest.java
index 0d4cab58..e1b0a27d 100644
---
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SessionTest.java
+++
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SessionTest.java
@@ -1579,4 +1579,95 @@ public class SessionTest extends
ImperativeClientTestCase {
peer.waitForScriptToComplete();
}
}
+
+ @Test
+ public void testNextReceiverRoundRobinPolicyReturnsCorrectReceiver()
throws Exception {
+ byte[] payload = createEncodedMessage(new AmqpValue<String>("Hello
World"));
+
+ try (ProtonTestServer peer = new ProtonTestServer()) {
+ peer.expectSASLAnonymousConnect();
+ peer.expectOpen().respond();
+ peer.expectBegin().respond();
+ peer.expectAttach().ofReceiver().respond();
+ peer.expectFlow().withLinkCredit(10);
+ peer.expectAttach().ofReceiver().respond();
+ peer.expectFlow().withLinkCredit(10);
+ peer.expectAttach().ofReceiver().respond();
+ peer.expectFlow().withLinkCredit(10);
+ peer.expectAttach().ofReceiver().respond();
+ peer.expectFlow().withLinkCredit(10);
+ peer.remoteTransfer().withHandle(1)
+ .withDeliveryId(0)
+ .withMore(false)
+ .withMessageFormat(0)
+ .withPayload(payload).queue();
+ peer.remoteTransfer().withHandle(1)
+ .withDeliveryId(1)
+ .withMore(false)
+ .withMessageFormat(0)
+ .withPayload(payload).queue();
+ peer.remoteTransfer().withHandle(2)
+ .withDeliveryId(2)
+ .withMore(false)
+ .withMessageFormat(0)
+ .withPayload(payload).queue();
+ peer.remoteTransfer().withHandle(3)
+ .withDeliveryId(3)
+ .withMore(false)
+ .withMessageFormat(0)
+ .withPayload(payload).queue();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Test started, peer listening on: {}", remoteURI);
+
+ Client container = Client.create();
+ ConnectionOptions options = new
ConnectionOptions().defaultNextReceiverPolicy(NextReceiverPolicy.RANDOM);
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort(), options);
+ SessionOptions sessionOptions = new
SessionOptions().defaultNextReceiverPolicy(NextReceiverPolicy.ROUND_ROBIN);
+ Session session = connection.openSession(sessionOptions);
+
+ ReceiverOptions receiverOptions = new
ReceiverOptions().creditWindow(10).autoAccept(false);
+ Receiver receiver1 = session.openReceiver("test-receiver1",
receiverOptions).openFuture().get();
+ Receiver receiver2 = session.openReceiver("test-receiver2",
receiverOptions).openFuture().get();
+ Receiver receiver3 = session.openReceiver("test-receiver3",
receiverOptions).openFuture().get();
+ Receiver receiver4 = session.openReceiver("test-receiver4",
receiverOptions).openFuture().get();
+
+ peer.waitForScriptToComplete();
+
+ Wait.waitFor(() -> receiver2.queuedDeliveries() == 2);
+ Wait.waitFor(() -> receiver3.queuedDeliveries() == 1);
+ Wait.waitFor(() -> receiver4.queuedDeliveries() == 1);
+
+ assertEquals(0, receiver1.queuedDeliveries());
+
+ Receiver next = session.nextReceiver();
+ assertSame(next, receiver2, "Should have gotten receiver 2");
+ next = session.nextReceiver();
+ assertSame(next, receiver3, "Should have gotten receiver 3");
+ next = session.nextReceiver();
+ assertSame(next, receiver4, "Should have gotten receiver 4");
+
+ peer.waitForScriptToComplete();
+
+ peer.remoteTransfer().withHandle(0)
+ .withDeliveryId(4)
+ .withMore(false)
+ .withMessageFormat(0)
+ .withPayload(payload).now();
+
+ Wait.waitFor(() -> receiver1.queuedDeliveries() == 1);
+
+ next = session.nextReceiver();
+ assertSame(next, receiver1);
+
+ peer.waitForScriptToComplete();
+ peer.expectClose().respond();
+
+ connection.close();
+
+ peer.waitForScriptToComplete();
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
