svn commit: r813962 - /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java

[rajdavies]

Author: rajdavies
Date: Fri Sep 11 18:06:11 2009
New Revision: 813962

URL: http://svn.apache.org/viewvc?rev=813962&view=rev
Log:
Apply patch from https://issues.apache.org/activemq/browse/AMQ-2002

Modified:
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java?rev=813962&r1=813961&r2=813962&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
 Fri Sep 11 18:06:11 2009
@@ -66,8 +66,8 @@
         }else {
             this.nonPersistent = new VMPendingMessageCursor();
         }
-        //TODO is this correct? we are ignoring the constructor parameter 
matchBatchSize
-//        this.nonPersistent.setMaxBatchSize(getMaxBatchSize());
+        
+        this.nonPersistent.setMaxBatchSize(maxBatchSize);
         this.nonPersistent.setSystemUsage(systemUsage);
         this.storePrefetches.add(this.nonPersistent);
     }
@@ -106,6 +106,7 @@
             tsp.setEnableAudit(isEnableAudit());
             tsp.setMaxAuditDepth(getMaxAuditDepth());
             tsp.setMaxProducersToAudit(getMaxProducersToAudit());
+            tsp.setMemoryUsageHighWaterMark(getMemoryUsageHighWaterMark());
             topics.put(destination, tsp);
             storePrefetches.add(tsp);
             if (isStarted()) {
@@ -259,6 +260,13 @@
         }
     }
     
+    public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark) {
+        super.setMemoryUsageHighWaterMark(memoryUsageHighWaterMark);
+        for (PendingMessageCursor cursor : storePrefetches) {
+            cursor.setMemoryUsageHighWaterMark(memoryUsageHighWaterMark);
+        }
+    }
+    
     public void setMaxProducersToAudit(int maxProducersToAudit) {
         super.setMaxProducersToAudit(maxProducersToAudit);
         for (PendingMessageCursor cursor : storePrefetches) {
@@ -304,7 +312,7 @@
         }
         return currentCursor;
     }
-
+    
     public String toString() {
         return "StoreDurableSubscriber(" + clientId + ":" + subscriberName + 
")";
     }