fix for http://issues.apache.org/activemq/browse/AMQ-857

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

activemq-core/src/main/java/org/apache/activemq/kaha/Store.java

@@ -141,4 +141,14 @@ public interface Store{
      * @throws IOException
      */
     public Set getListContainerIds() throws IOException;
+    
+    /**
+     * @return the maxDataFileLength
+     */
+    public long getMaxDataFileLength();
+
+    /**
+     * @param maxDataFileLength the maxDataFileLength to set
+     */
+    public void setMaxDataFileLength(long maxDataFileLength);
 }

activemq-core/src/main/java/org/apache/activemq/kaha/impl/DataManager.java

@@ -45,6 +45,7 @@ final class DataManager{
     private StoreDataReader reader;
     private StoreDataWriter writer;
     private DataFile currentWriteFile;
+    private long maxFileLength = MAX_FILE_LENGTH;
     Map fileMap=new HashMap();
 
     public static final int ITEM_HEAD_SIZE=5; // type + length
@@ -95,7 +96,7 @@ final class DataManager{
     }
 
     DataFile findSpaceForData(DataItem item) throws IOException{
-        if(currentWriteFile==null||((currentWriteFile.getLength()+item.getSize())>MAX_FILE_LENGTH)){
+        if(currentWriteFile==null||((currentWriteFile.getLength()+item.getSize())>maxFileLength)){
             int nextNum=currentWriteFile!=null?currentWriteFile.getNumber().intValue()+1:1;
             if(currentWriteFile!=null&&currentWriteFile.isUnused()){
                 removeDataFile(currentWriteFile);
@@ -260,4 +261,18 @@ final class DataManager{
     public void setRedoMarshaller(Marshaller redoMarshaller) {
         this.redoMarshaller = redoMarshaller;
     }
+
+    /**
+     * @return the maxFileLength
+     */
+    public long getMaxFileLength(){
+        return maxFileLength;
+    }
+
+    /**
+     * @param maxFileLength the maxFileLength to set
+     */
+    public void setMaxFileLength(long maxFileLength){
+        this.maxFileLength=maxFileLength;
+    }
 }

activemq-core/src/main/java/org/apache/activemq/kaha/impl/KahaStore.java

@@ -54,6 +54,7 @@ public class KahaStore implements Store{
     private String mode;
     private boolean initialized;
     private boolean logIndexChanges=false;
+    private long maxDataFileLength = DataManager.MAX_FILE_LENGTH;
 
     public KahaStore(String name,String mode) throws IOException{
         this.name=name;
@@ -279,6 +280,7 @@ public class KahaStore implements Store{
         DataManager dm = (DataManager) dataManagers.get(name);
         if (dm == null){
             dm = new DataManager(directory,name);
+            dm.setMaxFileLength(maxDataFileLength);
             recover(dm);
             dataManagers.put(name,dm);
         }
@@ -313,4 +315,18 @@ public class KahaStore implements Store{
         this.logIndexChanges = logIndexChanges;
     }
 
+    /**
+     * @return the maxDataFileLength
+     */
+    public long getMaxDataFileLength(){
+        return maxDataFileLength;
+    }
+
+    /**
+     * @param maxDataFileLength the maxDataFileLength to set
+     */
+    public void setMaxDataFileLength(long maxDataFileLength){
+        this.maxDataFileLength=maxDataFileLength;
+    }
+
 }

activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java

@@ -53,6 +53,7 @@ public class KahaPersistenceAdapter implements PersistenceAdapter{
     ConcurrentHashMap messageStores=new ConcurrentHashMap();
     private boolean useExternalMessageReferences;
     private OpenWireFormat wireFormat=new OpenWireFormat();
+    private long maxDataFileLength = 32 * 1024 * 1024;
     Store store;
 
     public KahaPersistenceAdapter(File dir) throws IOException{
@@ -61,6 +62,7 @@ public class KahaPersistenceAdapter implements PersistenceAdapter{
         }
         String name=dir.getAbsolutePath()+File.separator+"kaha.db";
         store=StoreFactory.open(name,"rw");
+        store.setMaxDataFileLength(maxDataFileLength);
     }
 
     public Set getDestinations(){
@@ -176,4 +178,18 @@ public class KahaPersistenceAdapter implements PersistenceAdapter{
      *            The UsageManager that is controlling the broker's memory usage.
      */
     public void setUsageManager(UsageManager usageManager){}
+
+    /**
+     * @return the maxDataFileLength
+     */
+    public long getMaxDataFileLength(){
+        return maxDataFileLength;
+    }
+
+    /**
+     * @param maxDataFileLength the maxDataFileLength to set
+     */
+    public void setMaxDataFileLength(long maxDataFileLength){
+        this.maxDataFileLength=maxDataFileLength;
+    }
 }