activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject svn commit: r591148 - in /activemq/trunk/activemq-core/src/main/java/org/apache/activemq: kaha/impl/async/AsyncDataManager.java kaha/impl/async/DataFile.java store/amq/AMQPersistenceAdapter.java
Date Thu, 01 Nov 2007 21:49:14 GMT
Author: rajdavies
Date: Thu Nov  1 14:49:13 2007
New Revision: 591148

URL: http://svn.apache.org/viewvc?rev=591148&view=rev
Log:
Fix for https://issues.apache.org/activemq/browse/AMQ-1486

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/DataFile.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java?rev=591148&r1=591147&r2=591148&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java
Thu Nov  1 14:49:13 2007
@@ -39,6 +39,7 @@
 import org.apache.activemq.kaha.impl.async.DataFileAppender.WriteKey;
 import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.IOHelper;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -63,8 +64,8 @@
 
     public static final byte DATA_ITEM_TYPE = 1;
     public static final byte REDO_ITEM_TYPE = 2;
-
     public static final String DEFAULT_DIRECTORY = "data";
+    public static final String DEFAULT_ARCHIVE_DIRECTORY = "data-archive";
     public static final String DEFAULT_FILE_PREFIX = "data-";
     public static final int DEFAULT_MAX_FILE_LENGTH = 1024 * 1024 * 32;
 
@@ -73,6 +74,7 @@
     protected final Map<WriteKey, WriteCommand> inflightWrites = new ConcurrentHashMap<WriteKey,
WriteCommand>();
 
     File directory = new File(DEFAULT_DIRECTORY);
+    File directoryArchive = new File (DEFAULT_ARCHIVE_DIRECTORY);
     String filePrefix = DEFAULT_FILE_PREFIX;
     ControlFile controlFile;
     boolean started;
@@ -91,6 +93,7 @@
     private final AtomicReference<Location> lastAppendLocation = new AtomicReference<Location>();
     private Runnable cleanupTask;
     private final AtomicLong storeSize;
+    private boolean archiveDataLogs;
     
     public AsyncDataManager(AtomicLong storeSize) {
         this.storeSize=storeSize;
@@ -409,15 +412,21 @@
         forceRemoveDataFile(dataFile);
     }
     
-    private synchronized void forceRemoveDataFile(DataFile dataFile) throws IOException {
+    private synchronized void forceRemoveDataFile(DataFile dataFile)
+            throws IOException {
         accessorPool.disposeDataFileAccessors(dataFile);
         DataFile removed = fileMap.remove(dataFile.getDataFileId());
         storeSize.addAndGet(-dataFile.getLength());
         dataFile.unlink();
-        boolean result = dataFile.delete();
-        LOG.debug("discarding data file " + dataFile
-                + (result ? "successful " : "failed"));
-
+        if (archiveDataLogs) {
+            dataFile.move(getDirectoryArchive());
+            LOG.debug("moced data file " + dataFile + " to "
+                    + getDirectoryArchive());
+        } else {
+            boolean result = dataFile.delete();
+            LOG.debug("discarding data file " + dataFile
+                    + (result ? "successful " : "failed"));
+        }
     }
 
     /**
@@ -580,8 +589,23 @@
 	public void setUseNio(boolean useNio) {
 		this.useNio = useNio;
 	}
-
 	
+	public File getDirectoryArchive() {
+        return directoryArchive;
+    }
+
+    public void setDirectoryArchive(File directoryArchive) {
+        this.directoryArchive = directoryArchive;
+    }
+    
+    public boolean isArchiveDataLogs() {
+        return archiveDataLogs;
+    }
+
+    public void setArchiveDataLogs(boolean archiveDataLogs) {
+        this.archiveDataLogs = archiveDataLogs;
+    }
+
     synchronized public Integer getCurrentDataFileId() {
         if( currentWriteFile==null )
             return null;

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/DataFile.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/DataFile.java?rev=591148&r1=591147&r2=591148&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/DataFile.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/DataFile.java
Thu Nov  1 14:49:13 2007
@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.io.RandomAccessFile;
 
+import org.apache.activemq.util.IOHelper;
 import org.apache.activemq.util.LinkedNode;
 
 /**
@@ -102,6 +103,10 @@
 
     public synchronized boolean delete() throws IOException {
         return file.delete();
+    }
+    
+    public synchronized void move(File targetDirectory) throws IOException{
+        IOHelper.moveFile(file,targetDirectory);
     }
 
     public int compareTo(DataFile df) {

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java?rev=591148&r1=591147&r2=591148&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
Thu Nov  1 14:49:13 2007
@@ -101,10 +101,12 @@
     private boolean syncOnWrite;
     private String brokerName = "";
     private File directory;
+    private File directoryArchive;
     private BrokerService brokerService;
     private AtomicLong storeSize = new AtomicLong();
     private boolean persistentIndex=true;
     private boolean useNio = true;
+    private boolean archiveDataLogs=false;
     private int maxFileLength = AsyncDataManager.DEFAULT_MAX_FILE_LENGTH;
     private int indexBinSize = HashIndex.DEFAULT_BIN_SIZE;
     private int indexKeySize = HashIndex.DEFAULT_KEY_SIZE;
@@ -143,8 +145,14 @@
                 this.directory = new File(directory, "amqstore");
             }
         }
+        if (this.directoryArchive == null) {
+            this.directoryArchive = new File(this.directory,"archive");
+        }
         LOG.info("AMQStore starting using directory: " + directory);
         this.directory.mkdirs();
+        if (archiveDataLogs) {
+            this.directoryArchive.mkdirs();
+        }
 
         if (this.usageManager != null) {
             this.usageManager.getMemoryUsage().addUsageListener(this);
@@ -624,6 +632,8 @@
     protected AsyncDataManager createAsyncDataManager() {
         AsyncDataManager manager = new AsyncDataManager(storeSize);
         manager.setDirectory(new File(directory, "journal"));
+        manager.setDirectoryArchive(getDirectoryArchive());
+        manager.setArchiveDataLogs(isArchiveDataLogs());
         manager.setMaxFileLength(maxFileLength);
         manager.setUseNio(useNio);    
         return manager;
@@ -793,6 +803,22 @@
     public void setIndexPageSize(int indexPageSize) {
         this.indexPageSize = indexPageSize;
     }
+    
+    public File getDirectoryArchive() {
+        return directoryArchive;
+    }
+
+    public void setDirectoryArchive(File directoryArchive) {
+        this.directoryArchive = directoryArchive;
+    }
+
+    public boolean isArchiveDataLogs() {
+        return archiveDataLogs;
+    }
+
+    public void setArchiveDataLogs(boolean archiveDataLogs) {
+        this.archiveDataLogs = archiveDataLogs;
+    }    
 
 	
 	protected void addInProgressDataFile(AMQMessageStore store,int dataFileId) {
@@ -809,5 +835,5 @@
         if (set != null) {
             set.remove(dataFileId);
         }
-    }    
+    }
 }



Mime
View raw message