You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2015/10/07 16:52:44 UTC
qpid-jms git commit: QPIDJMS-112: add option to only check the local
queue for messages when calling receive (when using prefetch),
allowing the old behaviour if desired
Repository: qpid-jms
Updated Branches:
refs/heads/master 162436223 -> 785682880
QPIDJMS-112: add option to only check the local queue for messages when calling receive (when using prefetch), allowing the old behaviour if desired
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/78568288
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/78568288
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/78568288
Branch: refs/heads/master
Commit: 785682880eefedd36cd7d2659f9dc02ffde0ba6c
Parents: 1624362
Author: Robert Gemmell <ro...@apache.org>
Authored: Wed Oct 7 15:30:07 2015 +0100
Committer: Robert Gemmell <ro...@apache.org>
Committed: Wed Oct 7 15:46:56 2015 +0100
----------------------------------------------------------------------
.../java/org/apache/qpid/jms/JmsConnection.java | 16 +++++
.../apache/qpid/jms/JmsConnectionFactory.java | 18 +++++
.../org/apache/qpid/jms/JmsMessageConsumer.java | 14 ++--
.../apache/qpid/jms/meta/JmsConnectionInfo.java | 18 +++++
.../integration/ConsumerIntegrationTest.java | 70 ++++++++++++++++++++
5 files changed, 129 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/78568288/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
index a395a12..e045b82 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnection.java
@@ -915,6 +915,22 @@ public class JmsConnection implements Connection, TopicConnection, QueueConnecti
connectionInfo.setRedeliveryPolicy(redeliveryPolicy);
}
+ public boolean isReceiveLocalOnly() {
+ return connectionInfo.isReceiveLocalOnly();
+ }
+
+ public void setReceiveLocalOnly(boolean receiveLocalOnly) {
+ this.connectionInfo.setReceiveLocalOnly(receiveLocalOnly);
+ }
+
+ public boolean isReceiveNoWaitLocalOnly() {
+ return connectionInfo.isReceiveNoWaitLocalOnly();
+ }
+
+ public void setReceiveNoWaitLocalOnly(boolean receiveNoWaitLocalOnly) {
+ this.connectionInfo.setReceiveNoWaitLocalOnly(receiveNoWaitLocalOnly);
+ }
+
public boolean isLocalMessagePriority() {
return connectionInfo.isLocalMessagePriority();
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/78568288/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
index ff0f2a9..8e7edfd 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionFactory.java
@@ -62,6 +62,8 @@ public class JmsConnectionFactory extends JNDIStorable implements ConnectionFact
private boolean sendAcksAsync;
private boolean localMessagePriority;
private boolean localMessageExpiry = true;
+ private boolean receiveLocalOnly;
+ private boolean receiveNoWaitLocalOnly;
private String queuePrefix = null;
private String topicPrefix = null;
private boolean validatePropertyNames = true;
@@ -705,4 +707,20 @@ public class JmsConnectionFactory extends JNDIStorable implements ConnectionFact
}
this.messageIDBuilder = messageIDBuilder;
}
+
+ public boolean isReceiveLocalOnly() {
+ return receiveLocalOnly;
+ }
+
+ public void setReceiveLocalOnly(boolean receiveLocalOnly) {
+ this.receiveLocalOnly = receiveLocalOnly;
+ }
+
+ public boolean isReceiveNoWaitLocalOnly() {
+ return receiveNoWaitLocalOnly;
+ }
+
+ public void setReceiveNoWaitLocalOnly(boolean receiveNoWaitLocalOnly) {
+ this.receiveNoWaitLocalOnly = receiveNoWaitLocalOnly;
+ }
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/78568288/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
index a793fa7..599d36c 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
@@ -227,7 +227,7 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
timeout = -1;
}
- return copy(ackFromReceive(dequeue(timeout)));
+ return copy(ackFromReceive(dequeue(timeout, connection.isReceiveLocalOnly())));
}
/**
@@ -240,7 +240,7 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
checkClosed();
checkMessageListener();
- return copy(ackFromReceive(dequeue(0)));
+ return copy(ackFromReceive(dequeue(0, connection.isReceiveNoWaitLocalOnly())));
}
/**
@@ -253,11 +253,12 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
*
* This method may consume messages that are expired or exceed a configured
* delivery count value but will continue to wait for the configured timeout.
- *
+ * @param localCheckOnly
+ * if false, try pulling a message if a >= 0 timeout expires with no message arriving
* @throws JMSException
* @return null if we timeout or if the consumer is closed concurrently.
*/
- private JmsInboundMessageDispatch dequeue(long timeout) throws JMSException {
+ private JmsInboundMessageDispatch dequeue(long timeout, boolean localCheckOnly) throws JMSException {
boolean pullConsumer = isPullConsumer();
boolean pullForced = pullConsumer;
@@ -284,14 +285,13 @@ public class JmsMessageConsumer implements MessageConsumer, JmsMessageAvailableC
}
if (envelope == null) {
- if ((timeout == 0 && pullForced) || pullConsumer || messageQueue.isClosed()) {
+ if ((timeout == 0 && (pullForced || localCheckOnly)) || pullConsumer || messageQueue.isClosed()) {
return null;
} else if (timeout > 0) {
timeout = Math.max(deadline - System.currentTimeMillis(), 0);
}
- //TODO: make it optional/configurable not to do this at all?
- if (timeout >= 0) {
+ if (timeout >= 0 && !localCheckOnly) {
// We don't do this for receive with no timeout since it
// is redundant: zero-prefetch consumers already pull, and
// the rest block indefinitely on the local messageQueue.
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/78568288/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
index 5adcb17..0b0e307 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/meta/JmsConnectionInfo.java
@@ -44,6 +44,8 @@ public final class JmsConnectionInfo implements JmsResource, Comparable<JmsConne
private boolean forceAsyncSend;
private boolean alwaysSyncSend;
private boolean validatePropertyNames = true;
+ private boolean receiveLocalOnly;
+ private boolean receiveNoWaitLocalOnly;
private boolean localMessagePriority;
private boolean localMessageExpiry;
private boolean sendAcksAsync;
@@ -219,6 +221,22 @@ public final class JmsConnectionInfo implements JmsResource, Comparable<JmsConne
this.sendAcksAsync = sendAcksAsync;
}
+ public boolean isReceiveLocalOnly() {
+ return receiveLocalOnly;
+ }
+
+ public void setReceiveLocalOnly(boolean receiveLocalOnly) {
+ this.receiveLocalOnly = receiveLocalOnly;
+ }
+
+ public boolean isReceiveNoWaitLocalOnly() {
+ return receiveNoWaitLocalOnly;
+ }
+
+ public void setReceiveNoWaitLocalOnly(boolean receiveNoWaitLocalOnly) {
+ this.receiveNoWaitLocalOnly = receiveNoWaitLocalOnly;
+ }
+
public boolean isLocalMessageExpiry() {
return localMessageExpiry;
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/78568288/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 4dd66c5..3df5b3a 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
@@ -43,6 +43,7 @@ import javax.jms.Topic;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsDefaultConnectionListener;
+import org.apache.qpid.jms.JmsPrefetchPolicy;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.apache.qpid.jms.test.QpidJmsTestCase;
import org.apache.qpid.jms.test.Wait;
@@ -567,4 +568,73 @@ public class ConsumerIntegrationTest extends QpidJmsTestCase {
testPeer.waitForAllHandlersToComplete(2000);
}
}
+
+ @Test(timeout=30000)
+ public void testReceiveWithTimoutDrainsOnNoMessage() throws IOException, Exception {
+ doDrainOnNoMessageTestImpl(false, true);
+ }
+
+ @Test(timeout=30000)
+ public void testReceiveWithTimoutDoesntDrainOnNoMessageWithLocalOnlyOption() throws IOException, Exception {
+ doDrainOnNoMessageTestImpl(true, true);
+ }
+
+ @Test(timeout=30000)
+ public void testReceiveNoWaitDrainsOnNoMessage() throws IOException, Exception {
+ doDrainOnNoMessageTestImpl(false, false);
+ }
+
+ @Test(timeout=30000)
+ public void testReceiveNoWaitDoesntDrainOnNoMessageWithLocalOnlyOption() throws IOException, Exception {
+ doDrainOnNoMessageTestImpl(true, false);
+ }
+
+ private void doDrainOnNoMessageTestImpl(boolean localCheckOnly, boolean noWait) throws JMSException, InterruptedException, Exception, IOException {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+ Connection connection = null;
+ if (localCheckOnly) {
+ if (noWait) {
+ connection = testFixture.establishConnecton(testPeer, "?jms.receiveNoWaitLocalOnly=true");
+ } else {
+ connection = testFixture.establishConnecton(testPeer, "?jms.receiveLocalOnly=true");
+ }
+ } else {
+ connection = testFixture.establishConnecton(testPeer);
+ }
+
+ connection.start();
+
+ testPeer.expectBegin();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+
+ // Expect receiver link attach and send credit
+ testPeer.expectReceiverAttach();
+ testPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(JmsPrefetchPolicy.DEFAULT_QUEUE_PREFETCH)));
+ if (!localCheckOnly) {
+ // If not doing local-check only, expect the credit to be drained
+ // and then replenished to open the window again
+ testPeer.expectLinkFlow(true, true, equalTo(UnsignedInteger.valueOf(JmsPrefetchPolicy.DEFAULT_QUEUE_PREFETCH)));
+ testPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(JmsPrefetchPolicy.DEFAULT_QUEUE_PREFETCH)));
+ }
+
+ MessageConsumer consumer = session.createConsumer(queue);
+ Message msg = null;
+ if (noWait) {
+ msg = consumer.receiveNoWait();
+ } else {
+ msg = consumer.receive(1);
+ }
+
+ assertNull(msg);
+
+ // Then close the consumer
+ testPeer.expectDetach(true, true, true);
+
+ consumer.close();
+
+ testPeer.waitForAllHandlersToComplete(3000);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org