QPID-1289: Make 0-8/0-9 client honour the max_preftech system property.

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

4 files changed, 9 insertions, 16 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
index ebeb29af78..f3865bd5af 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
@@ -250,7 +250,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
     protected AMQConnectionDelegate _delegate;
 
     // this connection maximum number of prefetched messages
-    private long _maxPrefetch;
+    protected int _maxPrefetch;
 
     //Indicates whether persistent messages are synchronized
     private boolean _syncPersistence;
@@ -337,13 +337,13 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         // set this connection maxPrefetch
         if (connectionURL.getOption(ConnectionURL.AMQ_MAXPREFETCH) != null)
         {
-            _maxPrefetch = Long.parseLong(connectionURL.getOption(ConnectionURL.AMQ_MAXPREFETCH));
+            _maxPrefetch = Integer.parseInt(connectionURL.getOption(ConnectionURL.AMQ_MAXPREFETCH));
         }
         else
         {
             // use the defaul value set for all connections
-            _maxPrefetch = Long.valueOf(System.getProperties().getProperty(ClientProperties.MAX_PREFETCH_PROP_NAME,
-                                                                           ClientProperties.MAX_PREFETCH_DEFAULT));
+            _maxPrefetch = Integer.parseInt(System.getProperties().getProperty(ClientProperties.MAX_PREFETCH_PROP_NAME,
+                                                                               ClientProperties.MAX_PREFETCH_DEFAULT));
         }
 
         if (connectionURL.getOption(ConnectionURL.AMQ_SYNC_PERSISTENCE) != null)
@@ -653,7 +653,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
     public org.apache.qpid.jms.Session createSession(final boolean transacted, final int acknowledgeMode) throws JMSException
     {
-        return createSession(transacted, acknowledgeMode, AMQSession.DEFAULT_PREFETCH_HIGH_MARK);
+        return createSession(transacted, acknowledgeMode, _maxPrefetch);
     }
 
     public org.apache.qpid.jms.Session createSession(final boolean transacted, final int acknowledgeMode, final int prefetch)
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 39f2ad5048..cbdefd0548 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
@@ -181,13 +181,6 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
     /** Used for debugging. */
     private static final Logger _logger = LoggerFactory.getLogger(AMQSession.class);
 
-
-    /** The default maximum number of prefetched message at which to suspend the channel. */
-    public static final int DEFAULT_PREFETCH_HIGH_MARK = 5000;
-
-    /** The default minimum number of prefetched messages at which to resume the channel. */
-    public static final int DEFAULT_PREFETCH_LOW_MARK = 2500;
-
     /**
      * The default value for immediate flag used by producers created by this session is false. That is, a consumer does
      * not need to be attached to a queue.
@@ -233,10 +226,10 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
     private int _ticket;
 
     /** Holds the high mark for prefetched message, at which the session is suspended. */
-    private int _defaultPrefetchHighMark = DEFAULT_PREFETCH_HIGH_MARK;
+    private int _defaultPrefetchHighMark;
 
     /** Holds the low mark for prefetched messages, below which the session is resumed. */
-    private int _defaultPrefetchLowMark = DEFAULT_PREFETCH_LOW_MARK;
+    private int _defaultPrefetchLowMark;
 
     /** Holds the message listener, if any, which is attached to this session. */
     private MessageListener _messageListener = null;
diff --git a/java/client/src/main/java/org/apache/qpid/client/XAConnectionImpl.java b/java/client/src/main/java/org/apache/qpid/client/XAConnectionImpl.java
index 476536e42b..20fa68605a 100644
--- a/java/client/src/main/java/org/apache/qpid/client/XAConnectionImpl.java
+++ b/java/client/src/main/java/org/apache/qpid/client/XAConnectionImpl.java
@@ -47,7 +47,7 @@ public class XAConnectionImpl extends AMQConnection implements XAConnection, XAQ
     public synchronized XASession createXASession() throws JMSException
     {
         checkNotClosed();
-        return _delegate.createXASession(AMQSession.DEFAULT_PREFETCH_HIGH_MARK, AMQSession.DEFAULT_PREFETCH_HIGH_MARK);
+        return _delegate.createXASession(_maxPrefetch, _maxPrefetch / 2);
     }
 
     //-- Interface  XAQueueConnection
diff --git a/java/client/src/main/java/org/apache/qpid/client/configuration/ClientProperties.java b/java/client/src/main/java/org/apache/qpid/client/configuration/ClientProperties.java
index edda18c715..49ac89d9b3 100644
--- a/java/client/src/main/java/org/apache/qpid/client/configuration/ClientProperties.java
+++ b/java/client/src/main/java/org/apache/qpid/client/configuration/ClientProperties.java
@@ -39,7 +39,7 @@ public class ClientProperties
      * type: long
      */
     public static final String MAX_PREFETCH_PROP_NAME = "max_prefetch";
-    public static final String MAX_PREFETCH_DEFAULT = "1000";
+    public static final String MAX_PREFETCH_DEFAULT = "5000";
 
     /**
      * When true a sync command is sent after every persistent messages.