diff options
Diffstat (limited to 'java/client/src')
3 files changed, 26 insertions, 7 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 eb161ee423..89ec7a8029 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 @@ -265,10 +265,10 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess * * @todo This is accessed only within a synchronized method, so does not need to be atomic. */ - private final AtomicBoolean _firstDispatcher = new AtomicBoolean(true); + protected final AtomicBoolean _firstDispatcher = new AtomicBoolean(true); /** Used to indicate that the session should start pre-fetching messages as soon as it is started. */ - private final boolean _immediatePrefetch; + protected final boolean _immediatePrefetch; /** Indicates that warnings should be generated on violations of the strict AMQP. */ private final boolean _strictAMQP; diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java index 3456124185..41efb4b586 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java @@ -528,4 +528,26 @@ public class AMQSession_0_10 extends AMQSession startDistpatcherIfNecessary(); } } + + synchronized void startDistpatcherIfNecessary() + { + // If IMMEDIATE_PREFETCH is not set then we need to start fetching + if (!_immediatePrefetch) + { + // We do this now if this is the first call on a started connection + if (isSuspended() && _firstDispatcher.getAndSet(false)) + { + try + { + suspendChannel(false); + } + catch (AMQException e) + { + _logger.info("Unsuspending channel threw an exception:" + e); + } + } + } + + startDistpatcherIfNecessary(false); + } } diff --git a/java/client/src/main/java/org/apache/qpidity/nclient/impl/ClientSession.java b/java/client/src/main/java/org/apache/qpidity/nclient/impl/ClientSession.java index 7883e995dd..dd2481368f 100644 --- a/java/client/src/main/java/org/apache/qpidity/nclient/impl/ClientSession.java +++ b/java/client/src/main/java/org/apache/qpidity/nclient/impl/ClientSession.java @@ -27,12 +27,9 @@ public class ClientSession extends org.apache.qpidity.transport.Session implemen { for (Range range : ranges) { - for (long l = range.getLower(); l <= range.getUpper(); l++) - { - System.out.println("Acknowleding transfer id : " + l); - super.processed(l); - } + super.processed(range); } + super.flushProcessed(); } public void messageSubscribe(String queue, String destination, short confirmMode, short acquireMode, MessagePartListener listener, Map<String, Object> filter, Option... options) |