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 9988d2e4 PROTON-2996 Handle remote close for default sender and session
9988d2e4 is described below
commit 9988d2e46c116634c62d2cfa891cfcbcbb8f683c
Author: Timothy Bish <[email protected]>
AuthorDate: Mon Nov 2 15:33:07 2026 -0500
PROTON-2996 Handle remote close for default sender and session
When the Connection default sender or session is remotely closed the
connection should reset the state such that a future call to use the
default sender or session recreates them and returns the new instance
---
.../protonj2/client/impl/ClientConnection.java | 13 ++
.../qpid/protonj2/client/impl/ClientLinkType.java | 12 +-
.../qpid/protonj2/client/impl/ClientSession.java | 13 ++
.../qpid/protonj2/client/impl/ConnectionTest.java | 86 ++++++++++++
.../qpid/protonj2/client/impl/SenderTest.java | 149 +++++++++++++++++++++
5 files changed, 267 insertions(+), 6 deletions(-)
diff --git
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
index 03cc362f..6afd394e 100644
---
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
+++
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientConnection.java
@@ -932,6 +932,19 @@ public final class ClientConnection implements Connection {
private ClientSession lazyCreateConnectionSession() throws ClientException
{
if (connectionSession == null) {
connectionSession = sessionBuilder.session(null).open();
+ connectionSession.remotelyClosedHandler((session) -> {
+ try {
+ session.closeAsync();
+ } catch (Throwable ignore) {}
+
+ // Clear the old default session and sender, a lazy create
needs to construct a new sender.
+ connectionSession = null;
+
+ if (connectionSender != null) {
+ connectionSender.closeAsync();
+ connectionSender = null;
+ }
+ });
}
return connectionSession;
diff --git
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientLinkType.java
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientLinkType.java
index b2155942..13bbd998 100644
---
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientLinkType.java
+++
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientLinkType.java
@@ -355,17 +355,17 @@ public abstract class ClientLinkType<LinkType extends
Link<LinkType>,
linkSpecificRemoteCloseHandler();
if (link.isLocallyOpen()) {
- if (linkRemotelyClosedHandler != null) {
- try {
- linkRemotelyClosedHandler.accept(self());
- } catch (Throwable ignore) {}
- }
-
immediateLinkShutdown(ClientExceptionSupport.convertToLinkClosedException(
link.getRemoteCondition(), "Link remotely closed without
explanation from the remote"));
} else {
immediateLinkShutdown(failureCause);
}
+
+ if (linkRemotelyClosedHandler != null) {
+ try {
+ linkRemotelyClosedHandler.accept(self());
+ } catch (Throwable ignore) {}
+ }
}
protected final void handleParentEndpointClosed(ProtonType link) {
diff --git
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSession.java
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSession.java
index 0b812597..3ffb3f14 100644
---
a/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSession.java
+++
b/protonj2-client/src/main/java/org/apache/qpid/protonj2/client/impl/ClientSession.java
@@ -23,6 +23,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.function.Consumer;
import java.util.function.Supplier;
import org.apache.qpid.protonj2.client.ErrorCondition;
@@ -75,6 +76,7 @@ public class ClientSession implements Session {
private volatile int closed;
private volatile ClientException failureCause;
private ClientTransactionContext txnContext = NO_OP_TXN_CONTEXT;
+ protected Consumer<ClientSession> sessionRemotelyClosedHandler;
private org.apache.qpid.protonj2.engine.Session protonSession;
@@ -424,6 +426,11 @@ public class ClientSession implements Session {
//----- Internal API accessible for use within the package
+ final ClientSession remotelyClosedHandler(Consumer<ClientSession> handler)
{
+ this.sessionRemotelyClosedHandler = handler;
+ return this;
+ }
+
ClientSession open() {
protonSession.localOpenHandler(this::handleLocalOpen)
.localCloseHandler(this::handleLocalClose)
@@ -579,6 +586,12 @@ public class ClientSession implements Session {
} else {
immediateSessionShutdown(failureCause);
}
+
+ if (sessionRemotelyClosedHandler != null) {
+ try {
+ sessionRemotelyClosedHandler.accept(this);
+ } catch (Throwable ignore) {}
+ }
}
private void handleEngineShutdown(Engine engine) {
diff --git
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ConnectionTest.java
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ConnectionTest.java
index b2bea64a..03c695c6 100644
---
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ConnectionTest.java
+++
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/ConnectionTest.java
@@ -19,6 +19,7 @@ package org.apache.qpid.protonj2.client.impl;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -1960,6 +1961,91 @@ public class ConnectionTest extends
ImperativeClientTestCase {
}
}
+ @Test
+ public void testDefaultSessionRecreatedOnNextUseIfRemotelyClosed() throws
Exception {
+ try (ProtonTestServer peer = new
ProtonTestServer(testServerOptions())) {
+ peer.expectSASLAnonymousConnect();
+ peer.expectOpen().respond();
+ peer.expectBegin().respond();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Connect test started, peer listening on: {}", remoteURI);
+
+ Client container = Client.create();
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort(), connectionOptions());
+ connection.openFuture().get(10, TimeUnit.SECONDS);
+
+ final Session defaultSession1 = connection.defaultSession();
+ assertNotNull(defaultSession1);
+ defaultSession1.openFuture().get(10, TimeUnit.SECONDS);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectEnd();
+ peer.remoteEnd().now();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectBegin().respond();
+
+ final Session defaultSession2 = connection.defaultSession();
+ assertNotNull(defaultSession2);
+ defaultSession2.openFuture().get(10, TimeUnit.SECONDS);
+
+ assertNotSame(defaultSession1, defaultSession2);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectClose().respond();
+
+ connection.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void
testUserClosesDefaultSessionResultsInNewSessionWhenRemotelyClosed() throws
Exception {
+ try (ProtonTestServer peer = new
ProtonTestServer(testServerOptions())) {
+ peer.expectSASLAnonymousConnect();
+ peer.expectOpen().respond();
+ peer.expectBegin().respond();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Connect test started, peer listening on: {}", remoteURI);
+
+ Client container = Client.create();
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort(), connectionOptions());
+ connection.openFuture().get(10, TimeUnit.SECONDS);
+
+ final Session defaultSession1 = connection.defaultSession();
+ assertNotNull(defaultSession1);
+ defaultSession1.openFuture().get(10, TimeUnit.SECONDS);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectEnd().respond();
+
+ defaultSession1.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectBegin().respond();
+
+ final Session defaultSession2 = connection.defaultSession();
+ assertNotNull(defaultSession2);
+ defaultSession2.openFuture().get(10, TimeUnit.SECONDS);
+
+ assertNotSame(defaultSession1, defaultSession2);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectClose().respond();
+
+ connection.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ }
+ }
+
@Disabled("Disabled due to requirement of hard coded port")
@Test
public void testLocalPortOption() throws Exception {
diff --git
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
index 7d6f7d36..9386c7f8 100644
---
a/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
+++
b/protonj2-client/src/test/java/org/apache/qpid/protonj2/client/impl/SenderTest.java
@@ -23,6 +23,7 @@ import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -3033,4 +3034,152 @@ public class SenderTest extends
ImperativeClientTestCase {
peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
}
}
+
+ @Test
+ public void
testDefaultSenderRecreatedOnNextUseIfDefaultSessionRemotelyClosed() throws
Exception {
+ try (ProtonTestServer peer = new ProtonTestServer()) {
+ peer.expectSASLAnonymousConnect();
+
peer.expectOpen().withDesiredCapabilities(ClientConstants.ANONYMOUS_RELAY.toString())
+ .respond()
+
.withOfferedCapabilities(ClientConstants.ANONYMOUS_RELAY.toString());
+ peer.expectBegin().respond();
+ peer.expectAttach().ofSender().respond();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Connect test started, peer listening on: {}", remoteURI);
+
+ Client container = Client.create();
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort());
+ connection.openFuture().get(10, TimeUnit.SECONDS);
+
+ final Sender defaultSender1 = connection.defaultSender();
+
+ assertNotNull(defaultSender1);
+
+ defaultSender1.openFuture().get(10, TimeUnit.SECONDS);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectEnd();
+ peer.remoteEnd().now();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectBegin().respond();
+ peer.expectAttach().ofSender().respond();
+
+ final Sender defaultSender2 = connection.defaultSender();
+ assertNotNull(defaultSender2);
+ defaultSender2.openFuture().get(10, TimeUnit.SECONDS);
+
+ assertNotNull(defaultSender1);
+ assertNotSame(defaultSender1, defaultSender2);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectClose().respond();
+
+ connection.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testDefaultSenderRecreatedWhenUserClosesDefaultSession()
throws Exception {
+ try (ProtonTestServer peer = new ProtonTestServer()) {
+ peer.expectSASLAnonymousConnect();
+
peer.expectOpen().withDesiredCapabilities(ClientConstants.ANONYMOUS_RELAY.toString())
+ .respond()
+
.withOfferedCapabilities(ClientConstants.ANONYMOUS_RELAY.toString());
+ peer.expectBegin().respond();
+ peer.expectAttach().ofSender().respond();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Connect test started, peer listening on: {}", remoteURI);
+
+ Client container = Client.create();
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort());
+ connection.openFuture().get(10, TimeUnit.SECONDS);
+
+ final Sender defaultSender1 = connection.defaultSender();
+
+ assertNotNull(defaultSender1);
+
+ defaultSender1.openFuture().get(10, TimeUnit.SECONDS);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectEnd().respond();
+
+ connection.defaultSession().close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectBegin().respond();
+ peer.expectAttach().ofSender().respond();
+
+ final Sender defaultSender2 = connection.defaultSender();
+ assertNotNull(defaultSender2);
+ defaultSender2.openFuture().get(10, TimeUnit.SECONDS);
+
+ assertNotNull(defaultSender1);
+ assertNotSame(defaultSender1, defaultSender2);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectClose().respond();
+
+ connection.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void
testUserClosesDefaultSenderResultsInNewSenderWhenRemotelyClosed() throws
Exception {
+ try (ProtonTestServer peer = new ProtonTestServer()) {
+ peer.expectSASLAnonymousConnect();
+
peer.expectOpen().withDesiredCapabilities(ClientConstants.ANONYMOUS_RELAY.toString())
+ .respond()
+
.withOfferedCapabilities(ClientConstants.ANONYMOUS_RELAY.toString());
+ peer.expectBegin().respond();
+ peer.expectAttach().ofSender().respond();
+ peer.start();
+
+ URI remoteURI = peer.getServerURI();
+
+ LOG.info("Connect test started, peer listening on: {}", remoteURI);
+
+ Client container = Client.create();
+ Connection connection = container.connect(remoteURI.getHost(),
remoteURI.getPort());
+ connection.openFuture().get(10, TimeUnit.SECONDS);
+
+ final Sender defaultSender1 = connection.defaultSender();
+
+ assertNotNull(defaultSender1);
+
+ defaultSender1.openFuture().get(10, TimeUnit.SECONDS);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectDetach().respond();
+
+ defaultSender1.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectAttach().ofSender().respond();
+
+ final Sender defaultSender2 = connection.defaultSender();
+ assertNotNull(defaultSender2);
+ defaultSender2.openFuture().get(10, TimeUnit.SECONDS);
+
+ assertNotNull(defaultSender1);
+ assertNotSame(defaultSender1, defaultSender2);
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectClose().respond();
+
+ connection.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
