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 2016/08/24 17:19:09 UTC
qpid-jms git commit: QPIDJMS-201 close or detach on drain timeout
based on durable or not.
Repository: qpid-jms
Updated Branches:
refs/heads/master 638e31cf6 -> a80b23a05
QPIDJMS-201 close or detach on drain timeout based on durable or not.
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/a80b23a0
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/a80b23a0
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/a80b23a0
Branch: refs/heads/master
Commit: a80b23a05aecea7f41445248bc58c7fe958fe437
Parents: 638e31c
Author: Timothy Bish <ta...@gmail.com>
Authored: Wed Aug 24 13:18:53 2016 -0400
Committer: Timothy Bish <ta...@gmail.com>
Committed: Wed Aug 24 13:18:53 2016 -0400
----------------------------------------------------------------------
.../jms/provider/amqp/AmqpAbstractResource.java | 2 +-
.../qpid/jms/provider/amqp/AmqpConsumer.java | 2 +-
.../integration/ConsumerIntegrationTest.java | 44 ++++++++++++++++----
3 files changed, 37 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/a80b23a0/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpAbstractResource.java
----------------------------------------------------------------------
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 a7d243f..a8599ac 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
@@ -155,7 +155,7 @@ public abstract class AmqpAbstractResource<R extends JmsResource, E extends Endp
if (endpoint != null) {
// TODO: if this is a producer/consumer link then we may only be detached,
// rather than fully closed, and should respond appropriately.
- endpoint.close();
+ closeOrDetachEndpoint();
}
// Process the close now, so that child close operations see the correct state.
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/a80b23a0/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
index a162af3..c15fd08 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java
@@ -123,7 +123,7 @@ public class AmqpConsumer extends AmqpAbstractResource<JmsConsumerInfo, Receiver
Exception cause = new JmsOperationTimedOutException("Remote did not respond to a drain request in time");
locallyClosed(session.getProvider(), cause);
stopRequest.onFailure(cause);
- session.getProvider().pumpToProtonTransport(stopRequest);
+ session.getProvider().pumpToProtonTransport();
}
}, getDrainTimeout());
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/a80b23a0/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
index 4f59bc8..b940301 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ConsumerIntegrationTest.java
@@ -755,15 +755,25 @@ public class ConsumerIntegrationTest extends QpidJmsTestCase {
@Test(timeout=30000)
public void testReceiveWithTimoutAndNoDrainResponseFailsAfterTimeout() throws IOException, Exception {
- doDrainWithNoResponseOnNoMessageTestImpl(false);
+ doDrainWithNoResponseOnNoMessageTestImpl(false, false);
}
@Test(timeout=30000)
public void testReceiveNoWaitAndNoDrainResponseFailsAfterTimeout() throws IOException, Exception {
- doDrainWithNoResponseOnNoMessageTestImpl(true);
+ doDrainWithNoResponseOnNoMessageTestImpl(true, false);
}
- private void doDrainWithNoResponseOnNoMessageTestImpl(boolean noWait) throws JMSException, InterruptedException, Exception, IOException {
+ @Test(timeout=30000)
+ public void testDurableReceiveWithTimoutAndNoDrainResponseFailsAfterTimeout() throws IOException, Exception {
+ doDrainWithNoResponseOnNoMessageTestImpl(false, true);
+ }
+
+ @Test(timeout=30000)
+ public void testDurableReceiveNoWaitAndNoDrainResponseFailsAfterTimeout() throws IOException, Exception {
+ doDrainWithNoResponseOnNoMessageTestImpl(true, true);
+ }
+
+ private void doDrainWithNoResponseOnNoMessageTestImpl(boolean noWait, boolean durable) throws JMSException, InterruptedException, Exception, IOException {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
Connection connection = null;
connection = testFixture.establishConnecton(testPeer, "?amqp.drainTimeout=500");
@@ -773,19 +783,35 @@ public class ConsumerIntegrationTest extends QpidJmsTestCase {
testPeer.expectBegin();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = session.createQueue("myQueue");
+ Topic topic = session.createTopic("topic");
// Expect receiver link attach and send credit
- testPeer.expectReceiverAttach();
- testPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(JmsDefaultPrefetchPolicy.DEFAULT_QUEUE_PREFETCH)));
+ if (durable) {
+ testPeer.expectDurableSubscriberAttach(topic.getTopicName(), getTestName());
+ } else {
+ testPeer.expectReceiverAttach();
+ }
+
+ testPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(JmsDefaultPrefetchPolicy.DEFAULT_TOPIC_PREFETCH)));
// Expect drain but do not respond so that the consumer times out.
- testPeer.expectLinkFlow(true, false, equalTo(UnsignedInteger.valueOf(JmsDefaultPrefetchPolicy.DEFAULT_QUEUE_PREFETCH)));
+ testPeer.expectLinkFlow(true, false, equalTo(UnsignedInteger.valueOf(JmsDefaultPrefetchPolicy.DEFAULT_TOPIC_PREFETCH)));
// Consumer should close due to timed waiting for drain.
- testPeer.expectDetach(true, true, true);
+ if (durable) {
+ // TODO - Bug in Proton-J causes extra flow to be emitted on detach.
+ testPeer.expectLinkFlow(true, false, equalTo(UnsignedInteger.valueOf(JmsDefaultPrefetchPolicy.DEFAULT_TOPIC_PREFETCH)));
+ testPeer.expectDetach(false, true, true);
+ } else {
+ testPeer.expectDetach(true, true, true);
+ }
- final MessageConsumer consumer = session.createConsumer(queue);
+ final MessageConsumer consumer;
+ if (durable) {
+ consumer = session.createDurableSubscriber(topic, getTestName());
+ } else {
+ consumer = session.createConsumer(topic);
+ }
try {
if (noWait) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org