This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-jms.git
The following commit(s) were added to refs/heads/master by this push:
new cec6fa3 QPIDJMS-491 Close the proton connection when transport error
is actioned
cec6fa3 is described below
commit cec6fa36ff606ae00664b40c30e8c7777936a37c
Author: Timothy Bish <[email protected]>
AuthorDate: Fri Feb 21 11:45:55 2020 -0500
QPIDJMS-491 Close the proton connection when transport error is actioned
Ensure that when the transport error event is actioned we close the
proton connection locally and then later also check that on resource
close if the provider has failed we don't wait for a remote response
that is never going to come.
---
.../qpid/jms/provider/amqp/AmqpAbstractResource.java | 7 +++++--
.../org/apache/qpid/jms/provider/amqp/AmqpProvider.java | 16 +++++++++++++++-
.../jms/integration/TransactionsIntegrationTest.java | 1 +
3 files changed, 21 insertions(+), 3 deletions(-)
diff --git
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
index 4507959..246fcbc 100644
---
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
+++
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
@@ -87,9 +87,12 @@ public abstract class AmqpAbstractResource<R extends
JmsResource, E extends Endp
resourceInfo.setState(ResourceState.CLOSED);
- // If already closed signal success or else the caller might never get
notified.
+ // If the current state of the connection or the provider doesn't
match what we'd expect for
+ // a normal close chain of events then we can assume something else
went wrong or was triggered
+ // by the remote so we just close our end and cleanup.
if (getEndpoint().getLocalState() == EndpointState.CLOSED ||
- getEndpoint().getRemoteState() == EndpointState.CLOSED) {
+ getEndpoint().getRemoteState() == EndpointState.CLOSED ||
+ getParent().getProvider().isFailed()) {
// Remote already closed this resource, close locally and free.
if (getEndpoint().getLocalState() != EndpointState.CLOSED) {
diff --git
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
index 8880e3a..59aa383 100644
---
a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
+++
b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
@@ -1012,11 +1012,17 @@ public class AmqpProvider implements Provider,
TransportListener , AmqpResourceP
case TRANSPORT_ERROR:
// We handle authentication failure elsewhere, but in
doing so we close the transport
// head which would also get us here, so only action
this if auth succeeded.
- if(authenticator == null ||
(authenticator.isComplete() && authenticator.wasSuccessful())) {
+ if (authenticator == null ||
(authenticator.isComplete() && authenticator.wasSuccessful())) {
protonTransportErrorHandled = true;
ErrorCondition transportCondition =
protonTransport.getCondition();
String message =
extractTransportErrorMessage(transportCondition);
+ // Transport has failed, ensure that we see local
end of connection as closed
+ // so other shutdown processing doesn't mistakenly
assume we can still get a
+ // close from the remote.
+ protonConnection.setCondition(transportCondition);
+ protonConnection.close();
+
throw new ProviderFailedException(message);
}
break;
@@ -1444,6 +1450,14 @@ public class AmqpProvider implements Provider,
TransportListener , AmqpResourceP
return remoteURI;
}
+ public Throwable getFailureCause() {
+ return failureCause;
+ }
+
+ public boolean isFailed() {
+ return failureCause != null;
+ }
+
@Override
public List<URI> getAlternateURIs() {
List<URI> alternates = new ArrayList<>();
diff --git
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TransactionsIntegrationTest.java
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TransactionsIntegrationTest.java
index 5eeecdb..abfd4b5 100644
---
a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TransactionsIntegrationTest.java
+++
b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/TransactionsIntegrationTest.java
@@ -1688,6 +1688,7 @@ public class TransactionsIntegrationTest extends
QpidJmsTestCase {
}
}
+ @Repeat(repetitions = 1)
@Test(timeout=20000)
public void testTransactionDeclaredDispositionWithoutTxnId() throws
Exception {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
