[hornetq-commits] JBoss hornetq SVN: r11176 - in branches/HORNETQ-720_Replication: hornetq-journal/src/main/java/org/hornetq/core/journal/impl and 1 other directory.
do-not-reply at jboss.org
do-not-reply at jboss.org
Wed Aug 10 06:28:45 EDT 2011
Author: borges
Date: 2011-08-10 06:28:45 -0400 (Wed, 10 Aug 2011)
New Revision: 11176
Added:
branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/FileWrapperJournal.java
Removed:
branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
Log:
HORNETQ-720 Rename ReplicatingJournal (as there was a ReplicatEDJournal already)
Modified: branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-08-10 01:53:32 UTC (rev 11175)
+++ branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-08-10 10:28:45 UTC (rev 11176)
@@ -26,9 +26,9 @@
import org.hornetq.core.journal.Journal;
import org.hornetq.core.journal.JournalLoadInformation;
import org.hornetq.core.journal.SequentialFile;
+import org.hornetq.core.journal.impl.FileWrapperJournal;
import org.hornetq.core.journal.impl.JournalFile;
import org.hornetq.core.journal.impl.JournalImpl;
-import org.hornetq.core.journal.impl.ReplicatingJournal;
import org.hornetq.core.logging.Logger;
import org.hornetq.core.paging.Page;
import org.hornetq.core.paging.PagedMessage;
@@ -441,7 +441,7 @@
/**
* Reserves files (with the given fileID) in the specified journal, and places a
- * {@link ReplicatingJournal} in place to store messages while synchronization is going on.
+ * {@link FileWrapperJournal} in place to store messages while synchronization is going on.
* @param packet
* @throws Exception
*/
@@ -458,8 +458,8 @@
JournalFile current = journal.createFilesForRemoteSync(packet.getFileIds(), mapToFill);
current.getFile().open(1, false);
registerJournal(packet.getJournalContentType().typeByte,
- new ReplicatingJournal(current, storage.hasCallbackSupport()));
- }
+ new FileWrapperJournal(current, storage.hasCallbackSupport()));
+ }
// XXX HORNETQ-720 really need to do away with this once the method calls get stable.
private static JournalImpl assertJournalImpl(final Journal journalIf) throws HornetQException
Copied: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/FileWrapperJournal.java (from rev 11169, branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java)
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/FileWrapperJournal.java (rev 0)
+++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/FileWrapperJournal.java 2011-08-10 10:28:45 UTC (rev 11176)
@@ -0,0 +1,229 @@
+package org.hornetq.core.journal.impl;
+
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.hornetq.api.core.HornetQException;
+import org.hornetq.core.journal.EncodingSupport;
+import org.hornetq.core.journal.IOCompletion;
+import org.hornetq.core.journal.Journal;
+import org.hornetq.core.journal.JournalLoadInformation;
+import org.hornetq.core.journal.LoaderCallback;
+import org.hornetq.core.journal.PreparedTransactionInfo;
+import org.hornetq.core.journal.RecordInfo;
+import org.hornetq.core.journal.TransactionFailureCallback;
+import org.hornetq.core.journal.impl.dataformat.JournalAddRecord;
+import org.hornetq.core.journal.impl.dataformat.JournalInternalRecord;
+
+/**
+ * Journal used at a replicating backup server during the synchronization of data with the 'live'
+ * server. It just wraps a single {@link JournalFile}.
+ * <p>
+ * Its main purpose is to store the data as a Journal would, but without verifying records.
+ */
+public class FileWrapperJournal extends JournalBase implements Journal
+{
+ private final ReentrantLock lockAppend = new ReentrantLock();
+ // private final ReadWriteLock journalLock = new ReentrantReadWriteLock();
+
+ private final JournalFile currentFile;
+
+ /**
+ * @param file
+ */
+ public FileWrapperJournal(JournalFile file, boolean hasCallbackSupport)
+ {
+ super(hasCallbackSupport);
+ currentFile = file;
+ }
+
+ @Override
+ public void start() throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void stop() throws Exception
+ {
+ currentFile.getFile().close();
+ }
+
+ @Override
+ public boolean isStarted()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ // ------------------------
+
+ // ------------------------
+
+// private void readLockJournal()
+// {
+// journalLock.readLock().lock();
+// }
+//
+// private void readUnlockJournal()
+// {
+// journalLock.readLock().unlock();
+// }
+
+ @Override
+ public void appendAddRecord(long id, byte recordType, EncodingSupport record, boolean sync, IOCompletion callback)
+ throws Exception
+ {
+ JournalInternalRecord addRecord = new JournalAddRecord(true, id, recordType, record);
+
+ writeRecord(addRecord, sync, callback);
+ }
+
+ /**
+ * Write the record to the current file.
+ */
+ private void writeRecord(JournalInternalRecord encoder, boolean sync, IOCompletion callback) throws Exception
+ {
+
+
+ lockAppend.lock();
+ try
+ {
+ if (callback != null)
+ {
+ callback.storeLineUp();
+ }
+
+ encoder.setFileID(currentFile.getRecordID());
+
+ if (callback != null)
+ {
+ currentFile.getFile().write(encoder, sync, callback);
+ }
+ else
+ {
+ currentFile.getFile().write(encoder, sync);
+ }
+ }
+ finally
+ {
+ lockAppend.unlock();
+ }
+ }
+
+ @Override
+ public void appendDeleteRecord(long id, boolean sync, IOCompletion completionCallback) throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void appendDeleteRecordTransactional(long txID, long id, EncodingSupport record) throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void appendAddRecordTransactional(long txID, long id, byte recordType, EncodingSupport record)
+ throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void
+ appendUpdateRecord(long id, byte recordType, EncodingSupport record, boolean sync, IOCompletion callback)
+ throws Exception
+ {
+ JournalInternalRecord updateRecord = new JournalAddRecord(false, id, recordType, record);
+ writeRecord(updateRecord, sync, callback);
+ }
+
+ @Override
+ public void appendUpdateRecordTransactional(long txID, long id, byte recordType, EncodingSupport record)
+ throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void appendCommitRecord(long txID, boolean sync, IOCompletion callback, boolean lineUpContext)
+ throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void appendPrepareRecord(long txID, EncodingSupport transactionData, boolean sync, IOCompletion callback)
+ throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void appendRollbackRecord(long txID, boolean sync, IOCompletion callback) throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public JournalLoadInformation load(LoaderCallback reloadManager) throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public JournalLoadInformation loadInternalOnly() throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public void lineUpContex(IOCompletion callback)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JournalLoadInformation load(List<RecordInfo> committedRecords,
+ List<PreparedTransactionInfo> preparedTransactions, TransactionFailureCallback transactionFailure)
+ throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public int getAlignment() throws Exception
+ {
+ throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
+ }
+
+ @Override
+ public int getNumberOfRecords()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getUserVersion()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void perfBlast(int pages)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void runDirectJournalBlast() throws Exception
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JournalLoadInformation loadSyncOnly() throws Exception
+ {
+ throw new UnsupportedOperationException();
+ }
+}
Deleted: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java 2011-08-10 01:53:32 UTC (rev 11175)
+++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java 2011-08-10 10:28:45 UTC (rev 11176)
@@ -1,232 +0,0 @@
-package org.hornetq.core.journal.impl;
-
-import java.util.List;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import org.hornetq.api.core.HornetQException;
-import org.hornetq.core.journal.EncodingSupport;
-import org.hornetq.core.journal.IOCompletion;
-import org.hornetq.core.journal.Journal;
-import org.hornetq.core.journal.JournalLoadInformation;
-import org.hornetq.core.journal.LoaderCallback;
-import org.hornetq.core.journal.PreparedTransactionInfo;
-import org.hornetq.core.journal.RecordInfo;
-import org.hornetq.core.journal.TransactionFailureCallback;
-import org.hornetq.core.journal.impl.dataformat.JournalAddRecord;
-import org.hornetq.core.journal.impl.dataformat.JournalInternalRecord;
-
-/**
- * Journal used at a replicating backup server during the synchronization of data with the 'live'
- * server.
- * <p>
- * Its main purpose is to store the data as a Journal would, but without verifying records.
- */
-public class ReplicatingJournal extends JournalBase implements Journal
-{
- private final ReentrantLock lockAppend = new ReentrantLock();
- private final ReadWriteLock journalLock = new ReentrantReadWriteLock();
-
- private final JournalFile currentFile;
-
- /**
- * @param file
- */
- public ReplicatingJournal(JournalFile file, boolean hasCallbackSupport)
- {
- super(hasCallbackSupport);
- currentFile = file;
- }
-
- @Override
- public void start() throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void stop() throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public boolean isStarted()
- {
- throw new UnsupportedOperationException();
- }
-
- // ------------------------
-
- // ------------------------
-
- private void readLockJournal()
- {
- journalLock.readLock().lock();
- }
-
- private void readUnlockJournal()
- {
- journalLock.readLock().unlock();
- }
-
- @Override
- public void appendAddRecord(long id, byte recordType, EncodingSupport record, boolean sync, IOCompletion callback)
- throws Exception
- {
- JournalInternalRecord addRecord = new JournalAddRecord(true, id, recordType, record);
-
- writeRecord(addRecord, sync, callback);
-
- }
-
- /**
- * Write the record to the current file.
- */
- private void writeRecord(JournalInternalRecord encoder, boolean sync, IOCompletion callback) throws Exception
- {
-
-
- lockAppend.lock();
- try
- {
- if (callback != null)
- {
- callback.storeLineUp();
- }
-
- encoder.setFileID(currentFile.getRecordID());
-
- if (callback != null)
- {
- currentFile.getFile().write(encoder, sync, callback);
- }
- else
- {
- currentFile.getFile().write(encoder, sync);
- }
- }
- finally
- {
- lockAppend.unlock();
- }
- }
-
- @Override
- public void appendDeleteRecord(long id, boolean sync, IOCompletion completionCallback) throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void appendDeleteRecordTransactional(long txID, long id, EncodingSupport record) throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void appendAddRecordTransactional(long txID, long id, byte recordType, EncodingSupport record)
- throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void
- appendUpdateRecord(long id, byte recordType, EncodingSupport record, boolean sync, IOCompletion callback)
- throws Exception
- {
- JournalInternalRecord updateRecord = new JournalAddRecord(false, id, recordType, record);
- writeRecord(updateRecord, sync, callback);
- }
-
- @Override
- public void appendUpdateRecordTransactional(long txID, long id, byte recordType, EncodingSupport record)
- throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void appendCommitRecord(long txID, boolean sync, IOCompletion callback, boolean lineUpContext)
- throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void appendPrepareRecord(long txID, EncodingSupport transactionData, boolean sync, IOCompletion callback)
- throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void appendRollbackRecord(long txID, boolean sync, IOCompletion callback) throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public JournalLoadInformation load(LoaderCallback reloadManager) throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public JournalLoadInformation loadInternalOnly() throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public void lineUpContex(IOCompletion callback)
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public JournalLoadInformation load(List<RecordInfo> committedRecords,
- List<PreparedTransactionInfo> preparedTransactions, TransactionFailureCallback transactionFailure)
- throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public int getAlignment() throws Exception
- {
- throw new HornetQException(HornetQException.UNSUPPORTED_PACKET);
- }
-
- @Override
- public int getNumberOfRecords()
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public int getUserVersion()
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void perfBlast(int pages)
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void runDirectJournalBlast() throws Exception
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public JournalLoadInformation loadSyncOnly() throws Exception
- {
- throw new UnsupportedOperationException();
- }
-}
More information about the hornetq-commits
mailing list