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 2015/09/01 23:19:24 UTC
qpid-jms git commit: QPIDJMS-51 Ensure that the pull timeout 0 case
drains credit as expected. Have the QueueBrowser use a receiveNoWait now that
the pull mechanisms are sorted. Update the tests to reflect the change.
Repository: qpid-jms
Updated Branches:
refs/heads/master f15f812ea -> 0f8e7fe73
QPIDJMS-51 Ensure that the pull timeout 0 case drains credit as
expected. Have the QueueBrowser use a receiveNoWait now that the pull
mechanisms are sorted. Update the tests to reflect the change.
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/0f8e7fe7
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/0f8e7fe7
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/0f8e7fe7
Branch: refs/heads/master
Commit: 0f8e7fe73700485c4f31ec187f7a740fd4e20fe2
Parents: f15f812
Author: Timothy Bish <ta...@gmail.com>
Authored: Tue Sep 1 17:19:14 2015 -0400
Committer: Timothy Bish <ta...@gmail.com>
Committed: Tue Sep 1 17:19:14 2015 -0400
----------------------------------------------------------------------
.../org/apache/qpid/jms/JmsQueueBrowser.java | 2 +-
.../qpid/jms/provider/amqp/AmqpConsumer.java | 15 +++++++++++---
.../QueueBrowserIntegrationTest.java | 21 +++++++++++---------
.../qpid/jms/consumer/JmsQueueBrowserTest.java | 18 +++++++++--------
4 files changed, 35 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/0f8e7fe7/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsQueueBrowser.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsQueueBrowser.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsQueueBrowser.java
index 1cf076c..398d462 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsQueueBrowser.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsQueueBrowser.java
@@ -132,7 +132,7 @@ public class JmsQueueBrowser implements QueueBrowser, Enumeration<Message> {
if (next == null) {
try {
- next = consumer.receive(2000);
+ next = consumer.receiveNoWait();
} catch (JMSException e) {
LOG.warn("Error while receive the next message: {}", e.getMessage());
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/0f8e7fe7/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 2ffb94b..5cb402d 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
@@ -108,7 +108,7 @@ public class AmqpConsumer extends AmqpAbstractResource<JmsConsumerInfo, Receiver
* Starts the consumer by setting the link credit to the given prefetch value.
*/
public void start(AsyncResult request) {
- getEndpoint().flow(resource.getPrefetchSize());
+ sendFlowIfNeeded();
request.onSuccess();
}
@@ -369,7 +369,9 @@ public class AmqpConsumer extends AmqpAbstractResource<JmsConsumerInfo, Receiver
int currentCredit = getEndpoint().getCredit();
if (currentCredit <= resource.getPrefetchSize() * 0.2) {
- getEndpoint().flow(resource.getPrefetchSize() - currentCredit);
+ int newCredit = resource.getPrefetchSize() - currentCredit;
+ LOG.trace("Consumer {} granting additional credit: {}", getConsumerId(), newCredit);
+ getEndpoint().flow(newCredit);
}
}
@@ -416,7 +418,14 @@ public class AmqpConsumer extends AmqpAbstractResource<JmsConsumerInfo, Receiver
}
} else if (timeout == 0) {
pullRequest = new DrainingPullRequest();
- getEndpoint().drain(1);
+ // If we have no credit then we need to issue some so that we can
+ // try to fulfill the pull request, otherwise we want to drain down
+ // what is there to ensure we consume everything available.
+ if (getEndpoint().getCredit() == 0) {
+ getEndpoint().drain(1);
+ } else {
+ getEndpoint().drain(0);
+ }
} else if (timeout > 0) {
// We need to drain the credit if no message arrives. If that
// happens, processing completion of the drain attempt will signal
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/0f8e7fe7/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/QueueBrowserIntegrationTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/QueueBrowserIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/QueueBrowserIntegrationTest.java
index e179db2..d29c083 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/QueueBrowserIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/QueueBrowserIntegrationTest.java
@@ -148,9 +148,11 @@ public class QueueBrowserIntegrationTest extends QpidJmsTestCase {
Queue queue = session.createQueue("myQueue");
// Expected the browser to create a consumer and send credit.
+ // Prefetch of one sends flow on every received message so expect a new flow.
testPeer.expectQueueBrowserAttach();
+ testPeer.expectLinkFlow();
testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent,
- 1, false, false, Matchers.equalTo(UnsignedInteger.ONE), 1, true, false);
+ 1, true, true, Matchers.equalTo(UnsignedInteger.ONE), 1, true, false);
testPeer.expectLinkFlow();
testPeer.expectDetach(true, true, true);
@@ -184,10 +186,11 @@ public class QueueBrowserIntegrationTest extends QpidJmsTestCase {
// Expected the browser to create a consumer and send credit, once hasMoreElements
// is called a message that is received should be accepted when the session is in
- // Auto acknowledge mode.
+ // Auto acknowledge mode with a drain from the browser.
testPeer.expectQueueBrowserAttach();
+ testPeer.expectLinkFlow();
testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent,
- 1, false, false, Matchers.greaterThan(UnsignedInteger.ZERO), 1, true, false);
+ 1, true, true, Matchers.greaterThan(UnsignedInteger.ZERO), 1, true, false);
testPeer.expectDetach(true, true, true);
QueueBrowser browser = session.createBrowser(queue);
@@ -217,8 +220,9 @@ public class QueueBrowserIntegrationTest extends QpidJmsTestCase {
DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);
testPeer.expectQueueBrowserAttach();
+ testPeer.expectLinkFlow();
testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent,
- 1, false, false, Matchers.greaterThan(UnsignedInteger.ZERO), 1, true, false);
+ 1, true, true, Matchers.greaterThan(UnsignedInteger.ZERO), 1, true, false);
// First expect an unsettled 'declare' transfer to the txn coordinator, and
// reply with a declared disposition state containing the txnId.
@@ -230,6 +234,7 @@ public class QueueBrowserIntegrationTest extends QpidJmsTestCase {
// Expected the browser to create a consumer and send credit, once hasMoreElements
// is called a message that is received should be delivered when the session is in
// a transacted session, the browser should not participate and close when asked.
+ testPeer.expectLinkFlow();
testPeer.expectDetach(true, true, true);
QueueBrowser browser = session.createBrowser(queue);
@@ -261,8 +266,9 @@ public class QueueBrowserIntegrationTest extends QpidJmsTestCase {
// is called a message that is received should be accepted when the session is in
// Auto acknowledge mode.
testPeer.expectQueueBrowserAttach();
+ testPeer.expectLinkFlow();
testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent,
- 1, false, false, Matchers.greaterThan(UnsignedInteger.ZERO), 1, true, false);
+ 1, true, true, Matchers.greaterThan(UnsignedInteger.ZERO), 1, true, false);
testPeer.expectDetach(true, true, true);
QueueBrowser browser = session.createBrowser(queue);
@@ -325,7 +331,6 @@ public class QueueBrowserIntegrationTest extends QpidJmsTestCase {
// Expected the browser to create a consumer and send credit.
testPeer.expectQueueBrowserAttach();
- testPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(1)));
testPeer.expectLinkFlow(true, true, equalTo(UnsignedInteger.valueOf(1)));
testPeer.expectDetach(true, true, true);
@@ -358,16 +363,14 @@ public class QueueBrowserIntegrationTest extends QpidJmsTestCase {
// Expected the browser to create a consumer and send credit.
testPeer.expectQueueBrowserAttach();
- testPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(1)));
// After timeout the browser should drain and here we want to ensure that if a
// message arrives that we handle it correctly.
testPeer.expectLinkFlowRespondWithTransfer(
null, null, null, null, amqpValueNullContent, 1, true,
- false, equalTo(UnsignedInteger.valueOf(1)), 1, true, false);
+ true, equalTo(UnsignedInteger.valueOf(1)), 1, true, false);
// Next attempt should not get a message and trigger a false on hasMoreElemets.
- testPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(1)));
testPeer.expectLinkFlow(true, true, equalTo(UnsignedInteger.valueOf(1)));
testPeer.expectDetach(true, true, true);
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/0f8e7fe7/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsQueueBrowserTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsQueueBrowserTest.java b/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsQueueBrowserTest.java
index d9a13a3..0a0d8c7 100644
--- a/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsQueueBrowserTest.java
+++ b/qpid-jms-interop-tests/qpid-jms-activemq-tests/src/test/java/org/apache/qpid/jms/consumer/JmsQueueBrowserTest.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.Enumeration;
+import java.util.concurrent.TimeUnit;
import javax.jms.Message;
import javax.jms.MessageConsumer;
@@ -46,7 +47,7 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
protected static final Logger LOG = LoggerFactory.getLogger(JmsQueueBrowserTest.class);
- @Test(timeout = 60000)
+ @Test(timeout = 40000)
public void testCreateQueueBrowser() throws Exception {
connection = createAmqpConnection();
connection.start();
@@ -63,7 +64,7 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
assertEquals(0, proxy.getQueueSize());
}
- @Test(timeout = 60000)
+ @Test(timeout = 40000)
public void testNoMessagesBrowserHasNoElements() throws Exception {
connection = createAmqpConnection();
connection.start();
@@ -110,7 +111,7 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
assertTrue(msg instanceof TextMessage);
}
- @Test(timeout = 60000)
+ @Test(timeout = 40000)
public void testBrowseAllInQueue() throws Exception {
connection = createAmqpConnection();
connection.start();
@@ -132,12 +133,13 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
assertNotNull(msg);
LOG.debug("Recv: {}", msg);
count++;
+ TimeUnit.MILLISECONDS.sleep(50);
}
assertFalse(enumeration.hasMoreElements());
assertEquals(5, count);
}
- @Test(timeout = 60000)
+ @Test(timeout = 40000)
public void testBrowseAllInQueuePrefetchOne() throws Exception {
connection = createAmqpConnection();
@@ -168,7 +170,7 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
assertEquals(5, count);
}
- @Test(timeout = 60000)
+ @Test(timeout = 40000)
public void testBrowseAllInQueueZeroPrefetch() throws Exception {
connection = createAmqpConnection();
@@ -199,7 +201,7 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
assertEquals(5, count);
}
- @Test(timeout = 60000)
+ @Test(timeout = 40000)
public void testBrowseAllInQueueTxSession() throws Exception {
connection = createAmqpConnection();
connection.start();
@@ -226,7 +228,7 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
assertEquals(5, count);
}
- @Test(timeout = 60000)
+ @Test(timeout = 40000)
public void testQueueBrowserInTxSessionLeavesOtherWorkUnaffected() throws Exception {
connection = createAmqpConnection();
connection.start();
@@ -269,7 +271,7 @@ public class JmsQueueBrowserTest extends AmqpTestSupport {
assertEquals(10, proxy.getQueueSize());
}
- @Test(timeout = 90000)
+ @Test(timeout = 60000)
public void testBrowseAllInQueueSmallPrefetch() throws Exception {
connection = createAmqpConnection();
((JmsConnection) connection).getPrefetchPolicy().setQueueBrowserPrefetch(10);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org