Make maximum page size for queues configurable

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

activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java

@@ -39,6 +39,7 @@ public abstract class BaseDestination implements Destination {
     private int maxProducersToAudit=1024;
     private int maxAuditDepth=1;
     private boolean enableAudit=true;
+    private int maxPageSize=1000;
     protected final DestinationStatistics destinationStatistics = new DestinationStatistics();
     
     /**
@@ -153,6 +154,11 @@ public abstract class BaseDestination implements Destination {
             destinationStatistics.getProducers().getCount() != 0;
     }
     
+    public int getMaxPageSize() {
+        return maxPageSize;
+    }
 
-    
+    public void setMaxPageSize(int maxPageSize) {
+        this.maxPageSize = maxPageSize;
+    }      
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/Destination.java

@@ -84,4 +84,8 @@ public interface Destination extends Service {
     void setEnableAudit(boolean enableAudit);
     
     boolean isActive();   
+    
+    int getMaxPageSize();
+    
+    public void setMaxPageSize(int maxPageSize);
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java

@@ -171,4 +171,12 @@ public class DestinationFilter implements Destination {
     public boolean isActive() {
         return next.isActive();
     }
+
+    public int getMaxPageSize() {
+        return next.getMaxPageSize();
+    }
+
+    public void setMaxPageSize(int maxPageSize) {
+        next.setMaxPageSize(maxPageSize);
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java

@@ -70,7 +70,6 @@ import org.apache.commons.logging.LogFactory;
  * @version $Revision: 1.28 $
  */
 public class Queue extends BaseDestination implements Task {
-    private static int MAXIMUM_PAGE_SIZE  = 1000;
     private final Log log;
     private final List<Subscription> consumers = new ArrayList<Subscription>(50);
     private PendingMessageCursor messages;
@@ -974,7 +973,7 @@ public class Queue extends BaseDestination implements Task {
     }
 
     private List<MessageReference> buildList(boolean force) throws Exception {
-        final int toPageIn = MAXIMUM_PAGE_SIZE - pagedInMessages.size();
+        final int toPageIn = getMaxPageSize() - pagedInMessages.size();
         List<MessageReference> result = null;
         if ((force || !consumers.isEmpty()) && toPageIn > 0) {
             messages.setMaxBatchSize(toPageIn);

activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java

@@ -56,6 +56,7 @@ public class PolicyEntry extends DestinationMapEntry {
     private boolean enableAudit=true;
     private boolean producerFlowControl = true;
     private boolean optimizedDispatch=false;
+    private int maxPageSize=1000;
    
     public void configure(Broker broker,Queue queue) {
         if (dispatchPolicy != null) {
@@ -76,6 +77,7 @@ public class PolicyEntry extends DestinationMapEntry {
         queue.setEnableAudit(isEnableAudit());
         queue.setMaxAuditDepth(getMaxQueueAuditDepth());
         queue.setMaxProducersToAudit(getMaxProducersToAudit());
+        queue.setMaxPageSize(getMaxPageSize());
     }
 
     public void configure(Topic topic) {
@@ -96,6 +98,7 @@ public class PolicyEntry extends DestinationMapEntry {
         topic.setEnableAudit(isEnableAudit());
         topic.setMaxAuditDepth(getMaxAuditDepth());
         topic.setMaxProducersToAudit(getMaxProducersToAudit());
+        topic.setMaxPageSize(getMaxPageSize());
     }
 
     public void configure(Broker broker, SystemUsage memoryManager, TopicSubscription subscription) {
@@ -350,4 +353,12 @@ public class PolicyEntry extends DestinationMapEntry {
         this.optimizedDispatch = optimizedDispatch;
     }
     
+    public int getMaxPageSize() {
+        return maxPageSize;
+    }
+
+    public void setMaxPageSize(int maxPageSize) {
+        this.maxPageSize = maxPageSize;
+    }    
+
 }