diff options
| author | Martin Ritchie <ritchiem@apache.org> | 2006-10-03 15:11:24 +0000 |
|---|---|---|
| committer | Martin Ritchie <ritchiem@apache.org> | 2006-10-03 15:11:24 +0000 |
| commit | b8a6eceee94a1e36ba826b7e491f328e49fd4c04 (patch) | |
| tree | 3c055daade8067ecc166295699c8fda0364a3c29 /qpid/java/client/src | |
| parent | c5008ad7e692bfaaac6235dea731eec1c6098bd5 (diff) | |
| download | qpid-python-b8a6eceee94a1e36ba826b7e491f328e49fd4c04.tar.gz | |
client/* - Only Create a Threshold Listener if if the acknowledgeMode is NO_ACK
common/*/framing/* - White space changes from tabs to 4 spaces.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@452529 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/java/client/src')
| -rw-r--r-- | qpid/java/client/src/org/apache/qpid/client/AMQSession.java | 7 | ||||
| -rw-r--r-- | qpid/java/client/src/org/apache/qpid/client/util/FlowControllingBlockingQueue.java | 19 |
2 files changed, 20 insertions, 6 deletions
diff --git a/qpid/java/client/src/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/org/apache/qpid/client/AMQSession.java index 4768399036..3bc670e609 100644 --- a/qpid/java/client/src/org/apache/qpid/client/AMQSession.java +++ b/qpid/java/client/src/org/apache/qpid/client/AMQSession.java @@ -220,6 +220,8 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi _channelId = channelId; _messageFactoryRegistry = messageFactoryRegistry; _defaultPrefetch = defaultPrefetch; + if (_acknowledgeMode == NO_ACKNOWLEDGE) + { _queue = new FlowControllingBlockingQueue(_defaultPrefetch, new FlowControllingBlockingQueue.ThresholdListener() { @@ -241,6 +243,11 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi } } }); + } + else + { + _queue = new FlowControllingBlockingQueue(_defaultPrefetch,null); + } } AMQSession(AMQConnection con, int channelId, boolean transacted, int acknowledgeMode) diff --git a/qpid/java/client/src/org/apache/qpid/client/util/FlowControllingBlockingQueue.java b/qpid/java/client/src/org/apache/qpid/client/util/FlowControllingBlockingQueue.java index ad2ca7b731..89e6968e44 100644 --- a/qpid/java/client/src/org/apache/qpid/client/util/FlowControllingBlockingQueue.java +++ b/qpid/java/client/src/org/apache/qpid/client/util/FlowControllingBlockingQueue.java @@ -63,11 +63,14 @@ public class FlowControllingBlockingQueue public Object take() throws InterruptedException { Object o = _queue.take(); - synchronized (_listener) + if (_listener != null) { - if (--_count == (_flowControlThreshold - 1)) + synchronized(_listener) { - _listener.underThreshold(_count); + if (--_count == (_flowControlThreshold - 1)) + { + _listener.underThreshold(_count); + } } } return o; @@ -76,12 +79,16 @@ public class FlowControllingBlockingQueue public void add(Object o) { _queue.add(o); - synchronized (_listener) + if (_listener != null) { - if (++_count == _flowControlThreshold) + synchronized(_listener) { - _listener.aboveThreshold(_count); + if (++_count == _flowControlThreshold) + { + _listener.aboveThreshold(_count); + } } } } } + |