diff options
author | Rafael H. Schloming <rhs@apache.org> | 2009-02-19 15:33:14 +0000 |
---|---|---|
committer | Rafael H. Schloming <rhs@apache.org> | 2009-02-19 15:33:14 +0000 |
commit | cfbe262e4de0c648bf72c47e1b32c3461b49df23 (patch) | |
tree | b4c71956161eeeb352f081fa6910a725fc0e0629 | |
parent | 3452ccc779803371e6ba04a1a3b9bec0d973d11d (diff) | |
download | qpid-python-cfbe262e4de0c648bf72c47e1b32c3461b49df23.tar.gz |
QPID-1665: turn off ack delay if qpid.session.max_ack_delay is set to a non-positive value
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@745900 13f79535-47bb-0310-9956-ffa450edef68
-rw-r--r-- | qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java | 43 |
1 files changed, 26 insertions, 17 deletions
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java index 8ab8110dd4..a8487b04e9 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java @@ -88,20 +88,8 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic // a ref on the qpid connection protected org.apache.qpid.transport.Connection _qpidConnection; - private TimerTask flushTask = new TimerTask() - { - public void run() - { - try - { - flushAcknowledgments(); - } - catch (Throwable t) - { - _logger.error("error flushing acks", t); - } - } - }; + private long maxAckDelay = Long.getLong("qpid.session.max_ack_delay", 1000); + private TimerTask flushTask = null; private RangeSet unacked = new RangeSet(); private int unackedCount = 0; @@ -138,7 +126,25 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic { _qpidSession.txSelect(); } - timer.schedule(flushTask, new Date(), Long.getLong("qpid.session.max_ack_delay", 1000)); + + if (maxAckDelay > 0) + { + flushTask = new TimerTask() + { + public void run() + { + try + { + flushAcknowledgments(); + } + catch (Throwable t) + { + _logger.error("error flushing acks", t); + } + } + }; + timer.schedule(flushTask, new Date(), maxAckDelay); + } } /** @@ -222,7 +228,7 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic long prefetch = getAMQConnection().getMaxPrefetch(); - if (unackedCount >= prefetch/2) + if (unackedCount >= prefetch/2 || maxAckDelay <= 0) { flushAcknowledgments(); } @@ -296,7 +302,10 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic */ public void sendClose(long timeout) throws AMQException, FailoverException { - flushTask.cancel(); + if (flushTask != null) + { + flushTask.cancel(); + } flushAcknowledgments(); getQpidSession().sync(); getQpidSession().close(); |