You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2014/10/21 18:31:48 UTC
git commit: Fix the case where unsubscribe fails and then is
reattempted and can succeed.
Repository: qpid-jms
Updated Branches:
refs/heads/master f1584d25c -> cc1559be5
Fix the case where unsubscribe fails and then is reattempted and can
succeed.
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/cc1559be
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/cc1559be
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/cc1559be
Branch: refs/heads/master
Commit: cc1559be52bba177ae44d35e6669d2222eb7c741
Parents: f1584d2
Author: Timothy Bish <ta...@gmail.com>
Authored: Tue Oct 21 12:31:34 2014 -0400
Committer: Timothy Bish <ta...@gmail.com>
Committed: Tue Oct 21 12:31:34 2014 -0400
----------------------------------------------------------------------
.../qpid/jms/provider/amqp/AmqpConnectionSession.java | 11 ++++++++---
.../qpid/jms/consumer/JmsDurableSubscriberTest.java | 2 --
2 files changed, 8 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/cc1559be/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnectionSession.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnectionSession.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnectionSession.java
index f640e64..eaddc17 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnectionSession.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConnectionSession.java
@@ -109,19 +109,21 @@ public class AmqpConnectionSession extends AmqpSession {
@Override
public void onSuccess() {
- final Source returnedSource = (Source) requestor.getEndpoint().getRemoteSource();
- if (returnedSource == null) {
+ Object result = requestor.getEndpoint().getRemoteSource();
+ if (result == null || !(result instanceof Source)) {
LOG.trace("No Source returned for subscription: {}", requestor.getSubscriptionName());
pendingUnsubs.remove(requestor.getSubscriptionName());
+ requestor.closed();
super.onFailure(new IOException("Could not fetch remote subscription information"));
} else {
+ final Source remoteSource = (Source) result;
LOG.trace("Source returned for subscription: {} closing first stage", requestor.getSubscriptionName());
requestor.close(new AsyncResult() {
@Override
public void onSuccess() {
RemoveDurabilityRequestor removeRequestor =
- new RemoveDurabilityRequestor(getJmsResource(), requestor.getSubscriptionName(), returnedSource);
+ new RemoveDurabilityRequestor(getJmsResource(), requestor.getSubscriptionName(), remoteSource);
RemoveDurabilityRequest removeRequest = new RemoveDurabilityRequest(removeRequestor, getWrappedRequest());
pendingUnsubs.put(requestor.getSubscriptionName(), removeRequest);
LOG.trace("Second stage remove started for subscription: {}", requestor.getSubscriptionName());
@@ -146,6 +148,7 @@ public class AmqpConnectionSession extends AmqpSession {
@Override
public void onFailure(Throwable result) {
pendingUnsubs.remove(requestor.getSubscriptionName());
+ requestor.closed();
super.onFailure(result);
}
}
@@ -179,6 +182,7 @@ public class AmqpConnectionSession extends AmqpSession {
if (isAwaitingOpen()) {
openRequest.onSuccess();
} else {
+ closed();
AmqpConnectionSession.this.reportError(new IOException("Durable unsubscribe failed unexpectedly"));
}
}
@@ -212,6 +216,7 @@ public class AmqpConnectionSession extends AmqpSession {
public void onFailure(Throwable result) {
LOG.trace("Second stage remove failed for subscription: {}", requestor.getSubscriptionName());
pendingUnsubs.remove(requestor.getSubscriptionName());
+ requestor.closed();
super.onFailure(result);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/cc1559be/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsDurableSubscriberTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsDurableSubscriberTest.java b/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsDurableSubscriberTest.java
index 74ed558..d81b30b 100644
--- a/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsDurableSubscriberTest.java
+++ b/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsDurableSubscriberTest.java
@@ -36,7 +36,6 @@ import javax.jms.TopicSubscriber;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.TopicViewMBean;
import org.apache.qpid.jms.support.AmqpTestSupport;
-import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -136,7 +135,6 @@ public class JmsDurableSubscriberTest extends AmqpTestSupport {
assertEquals(0, broker.getInactiveDurableTopicSubscribers().length);
}
- @Ignore
@Test(timeout = 60000)
public void testDuableSubscriptionUnsubscribeInUseThrowsAndRecovers() throws Exception {
connection = createAmqpConnection();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org