Changed to send message ack when required

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@581912 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 27 insertions, 8 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)
diff --git a/java/common/src/main/java/org/apache/qpidity/transport/Session.java b/java/common/src/main/java/org/apache/qpidity/transport/Session.java
index ad2cbe8cde..5441ac8f23 100644
--- a/java/common/src/main/java/org/apache/qpidity/transport/Session.java
+++ b/java/common/src/main/java/org/apache/qpidity/transport/Session.java
@@ -111,7 +111,7 @@ public class Session extends Invoker
         }
     }
 
-    void flushProcessed()
+   public void flushProcessed()
     {
         long mark = -1;
         boolean first = true;