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/qpid@745900 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 26 insertions, 17 deletions

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 8ab8110dd4..a8487b04e9 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
@@ -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();