AMQ-6403 - add indexDirectory attribute to kahadb plist impl - settable via broker service tempDataStore

activemq-broker/src/test/java/org/apache/activemq/store/PListTestSupport.java

@@ -45,7 +45,7 @@ import org.slf4j.LoggerFactory;
 
 public abstract class PListTestSupport {
     static final Logger LOG = LoggerFactory.getLogger(PListTestSupport.class);
-    private PListStore store;
+    protected PListStore store;
     private PList plist;
     final ByteSequence payload = new ByteSequence(new byte[400]);
     final String idSeed = new String("Seed" + new byte[1024]);

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/plist/PListStoreImpl.java

@@ -54,6 +54,7 @@ public class PListStoreImpl extends ServiceSupport implements BrokerServiceAware
     static final int OPEN_STATE = 2;
 
     private File directory;
+    private File indexDirectory;
     PageFile pageFile;
     private Journal journal;
     private LockFile lockFile;
@@ -202,6 +203,14 @@ public class PListStoreImpl extends ServiceSupport implements BrokerServiceAware
         this.directory = directory;
     }
 
+    public File getIndexDirectory() {
+        return indexDirectory != null ? indexDirectory : directory;
+    }
+
+    public void setIndexDirectory(File indexDirectory) {
+        this.indexDirectory = indexDirectory;
+    }
+
     public long size() {
         synchronized (this) {
             if (!initialized) {
@@ -277,13 +286,17 @@ public class PListStoreImpl extends ServiceSupport implements BrokerServiceAware
                 }
                 IOHelper.mkdirs(this.directory);
                 IOHelper.deleteChildren(this.directory);
+                if (this.indexDirectory != null) {
+                    IOHelper.mkdirs(this.indexDirectory);
+                    IOHelper.deleteChildren(this.indexDirectory);
+                }
                 lock();
                 this.journal = new Journal();
                 this.journal.setDirectory(directory);
                 this.journal.setMaxFileLength(getJournalMaxFileLength());
                 this.journal.setWriteBatchSize(getJournalMaxWriteBatchSize());
                 this.journal.start();
-                this.pageFile = new PageFile(directory, "tmpDB");
+                this.pageFile = new PageFile(getIndexDirectory(), "tmpDB");
                 this.pageFile.setEnablePageCaching(getIndexEnablePageCaching());
                 this.pageFile.setPageSize(getIndexPageSize());
                 this.pageFile.setWriteBatchSize(getIndexWriteBatchSize());
@@ -485,6 +498,9 @@ public class PListStoreImpl extends ServiceSupport implements BrokerServiceAware
     @Override
     public String toString() {
         String path = getDirectory() != null ? getDirectory().getAbsolutePath() : "DIRECTORY_NOT_SET";
+        if (indexDirectory != null) {
+            path += "|" + indexDirectory.getAbsolutePath();
+        }
         return "PListStore:[" + path + "]";
     }
 }

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/plist/PListImplTest.java

@@ -18,6 +18,12 @@ package org.apache.activemq.store.kahadb.plist;
 
 import org.apache.activemq.store.PListStore;
 import org.apache.activemq.store.PListTestSupport;
+import org.junit.Test;
+
+import java.io.File;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
 
 /**
  * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
@@ -65,4 +71,23 @@ public class PListImplTest extends PListTestSupport {
         store.setIndexPageSize(2*1024);
         return store;
     }
+
+    @Test
+    public void testIndexDir() throws Exception {
+        PListStoreImpl pListStore = (PListStoreImpl)store;
+        assertEquals(pListStore.getDirectory(), pListStore.getIndexDirectory());
+    }
+
+    @Test
+    public void testSetIndexDir() throws Exception {
+        PListStoreImpl pListStore = (PListStoreImpl)store;
+        final File directory = pListStore.getDirectory();
+        pListStore.stop();
+        pListStore = createPListStore();
+        pListStore.setLazyInit(false);
+        pListStore.setIndexDirectory(new File(directory, "indexDir"));
+        pListStore.start();
+        assertNotEquals(pListStore.getDirectory(), pListStore.getIndexDirectory());
+        pListStore.stop();
+    }
 }