From e7c9d216600558bd097c9599c879ee8ef5328d87 Mon Sep 17 00:00:00 2001
From: Robert Davies <rajdavies@apache.org>
Date: Fri, 11 Sep 2009 18:06:11 +0000
Subject: [PATCH] Apply patch from
 https://issues.apache.org/activemq/browse/AMQ-2002

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@813962 13f79535-47bb-0310-9956-ffa450edef68
---
 .../cursors/StoreDurableSubscriberCursor.java      | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java b/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
index 2d5da2612d..410ab1d51a 100755
--- a/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
+++ b/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
@@ -66,8 +66,8 @@ public class StoreDurableSubscriberCursor extends AbstractPendingMessageCursor {
         }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 @@ public class StoreDurableSubscriberCursor extends AbstractPendingMessageCursor {
             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 class StoreDurableSubscriberCursor extends AbstractPendingMessageCursor {
         }
     }
     
+    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 @@ public class StoreDurableSubscriberCursor extends AbstractPendingMessageCursor {
         }
         return currentCursor;
     }
-
+    
     public String toString() {
         return "StoreDurableSubscriber(" + clientId + ":" + subscriberName + ")";
     }