diff options
author | Robert Godfrey <rgodfrey@apache.org> | 2012-01-03 19:48:46 +0000 |
---|---|---|
committer | Robert Godfrey <rgodfrey@apache.org> | 2012-01-03 19:48:46 +0000 |
commit | 02bbab932f5f845bfa8eac6069bc4159bbe53d07 (patch) | |
tree | 7a31c2804f9b9fa5f1dbabc80cbe219d9e735890 /java/client/src | |
parent | 7b0c33ff443deb937d26f07c039bd483e9bcbe29 (diff) | |
download | qpid-python-02bbab932f5f845bfa8eac6069bc4159bbe53d07.tar.gz |
QPID-3720 : [Java Broker] Implement Message Grouping
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1226930 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java/client/src')
3 files changed, 8 insertions, 9 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index 8984b7ca8c..4b6015945e 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -1583,6 +1583,11 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic return _prefetchLowMark; } + public int getPrefetch() + { + return _prefetchHighMark; + } + public AMQShortString getDefaultQueueExchangeName() { return _connection.getDefaultQueueExchangeName(); @@ -3047,7 +3052,7 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic */ public boolean prefetch() { - return getAMQConnection().getMaxPrefetch() > 0; + return _prefetchHighMark > 0; } /** Signifies that the session has pending sends to commit. */ diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java index a49e31ce8c..3b6179dd07 100644 --- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java +++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java @@ -545,7 +545,7 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<UnprocessedM } else if (getSession().prefetch()) { - capacity = _0_10session.getAMQConnection().getMaxPrefetch(); + capacity = getSession().getPrefetch(); } return capacity; } diff --git a/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java b/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java index f53fa8d83c..1889577773 100644 --- a/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java +++ b/java/client/src/test/java/org/apache/qpid/client/AMQSession_0_10Test.java @@ -588,7 +588,7 @@ public class AMQSession_0_10Test extends TestCase } boolean isTransacted = acknowledgeMode == javax.jms.Session.SESSION_TRANSACTED ? true : false; AMQSession_0_10 session = new AMQSession_0_10(createConnection(throwException), amqConnection, 1, isTransacted, acknowledgeMode, - 1, 1, "test"); + 0, 0, "test"); return session; } @@ -600,7 +600,6 @@ public class AMQSession_0_10Test extends TestCase connection.setSessionFactory(new SessionFactory() { - @Override public Session newSession(Connection conn, Binary name, long expiry) { return new MockSession(conn, new SessionDelegate(), name, expiry, throwException); @@ -611,7 +610,6 @@ public class AMQSession_0_10Test extends TestCase private final class MockMessageListener implements MessageListener { - @Override public void onMessage(Message arg0) { } @@ -710,23 +708,19 @@ public class AMQSession_0_10Test extends TestCase { private List<ProtocolEvent> _sendEvents = new ArrayList<ProtocolEvent>(); - @Override public void setIdleTimeout(int i) { } - @Override public void send(ProtocolEvent msg) { _sendEvents.add(msg); } - @Override public void flush() { } - @Override public void close() { } |