Author: borges Date: 2011-08-09 06:26:46 -0400 (Tue, 09 Aug 2011) New Revision: 11157 Added: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalBase.java Modified: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java Log: HORNETQ-720 Move common 'boiler-plate' journaling code into base-class Added: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalBase.java =================================================================== --- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalBase.java (rev 0) +++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalBase.java 2011-08-09 10:26:46 UTC (rev 11157) @@ -0,0 +1,216 @@ +package org.hornetq.core.journal.impl; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.journal.EncodingSupport; +import org.hornetq.core.journal.IOCompletion; +import org.hornetq.core.journal.impl.dataformat.ByteArrayEncoding; + +abstract class JournalBase +{ + + private final boolean hasCallbackSupport; + + public JournalBase(boolean hasCallbackSupport) + { + this.hasCallbackSupport = hasCallbackSupport; + } + + abstract public void appendAddRecord(final long id, final byte recordType, final EncodingSupport record, + final boolean sync, final IOCompletion callback) throws Exception; + + abstract public void appendAddRecordTransactional(final long txID, final long id, final byte recordType, + final EncodingSupport record) throws Exception; + + abstract public void appendCommitRecord(final long txID, final boolean sync, final IOCompletion callback, + boolean lineUpContext) throws Exception; + + abstract public void appendDeleteRecord(final long id, final boolean sync, final IOCompletion callback) + throws Exception; + + abstract public void appendDeleteRecordTransactional(final long txID, final long id, final EncodingSupport record) + throws Exception; + + abstract public void appendPrepareRecord(final long txID, final EncodingSupport transactionData, final boolean sync, + final IOCompletion callback) throws Exception; + + abstract public void appendUpdateRecord(final long id, final byte recordType, final EncodingSupport record, + final boolean sync, final IOCompletion callback) throws Exception; + + abstract public void appendUpdateRecordTransactional(final long txID, final long id, final byte recordType, + final EncodingSupport record) throws Exception; + + abstract public void appendRollbackRecord(final long txID, final boolean sync, final IOCompletion callback) + throws Exception; + + + public void appendAddRecord(long id, byte recordType, byte[] record, boolean sync) throws Exception + { + appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync); + } + + public void appendAddRecord(long id, byte recordType, byte[] record, boolean sync, IOCompletion completionCallback) + throws Exception + { + appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync, completionCallback); + } + + public void appendAddRecord(long id, byte recordType, EncodingSupport record, boolean sync) throws Exception + { + SyncIOCompletion callback = getSyncCallback(sync); + + appendAddRecord(id, recordType, record, sync, callback); + + if (callback != null) + { + callback.waitCompletion(); + } + } + + public void appendCommitRecord(final long txID, final boolean sync) throws Exception + { + SyncIOCompletion syncCompletion = getSyncCallback(sync); + + appendCommitRecord(txID, sync, syncCompletion, true); + + if (syncCompletion != null) + { + syncCompletion.waitCompletion(); + } + } + + public void appendCommitRecord(final long txID, final boolean sync, final IOCompletion callback) throws Exception + { + appendCommitRecord(txID, sync, callback, true); + } + public void appendUpdateRecord(final long id, final byte recordType, final byte[] record, final boolean sync) + throws Exception + { + appendUpdateRecord(id, recordType, new ByteArrayEncoding(record), sync); + } + + public void appendUpdateRecordTransactional(final long txID, final long id, final byte recordType, + final byte[] record) throws Exception + { + appendUpdateRecordTransactional(txID, id, recordType, new ByteArrayEncoding(record)); + } + + public void appendUpdateRecord(final long id, final byte recordType, final byte[] record, final boolean sync, + final IOCompletion callback) throws Exception + { + appendUpdateRecord(id, recordType, new ByteArrayEncoding(record), sync, callback); + } + + public void appendAddRecordTransactional(final long txID, final long id, final byte recordType, final byte[] record) + throws Exception + { + appendAddRecordTransactional(txID, id, recordType, new ByteArrayEncoding(record)); + } + + public void appendDeleteRecordTransactional(final long txID, final long id) throws Exception + { + appendDeleteRecordTransactional(txID, id, NullEncoding.instance); + } + + public void appendPrepareRecord(final long txID, final byte[] transactionData, final boolean sync, + final IOCompletion completion) throws Exception + { + appendPrepareRecord(txID, new ByteArrayEncoding(transactionData), sync, completion); + } + + public void appendPrepareRecord(final long txID, final byte[] transactionData, final boolean sync) throws Exception + { + appendPrepareRecord(txID, new ByteArrayEncoding(transactionData), sync); + } + + public void appendPrepareRecord(final long txID, final EncodingSupport transactionData, final boolean sync) + throws Exception + { + SyncIOCompletion syncCompletion = getSyncCallback(sync); + + appendPrepareRecord(txID, transactionData, sync, syncCompletion); + + if (syncCompletion != null) + { + syncCompletion.waitCompletion(); + } + } + + public void appendDeleteRecordTransactional(final long txID, final long id, final byte[] record) throws Exception + { + appendDeleteRecordTransactional(txID, id, new ByteArrayEncoding(record)); + } + + public void + appendUpdateRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync) + throws Exception + { + SyncIOCompletion callback = getSyncCallback(sync); + + appendUpdateRecord(id, recordType, record, sync, callback); + + if (callback != null) + { + callback.waitCompletion(); + } + } + + public void appendRollbackRecord(final long txID, final boolean sync) throws Exception + { + SyncIOCompletion syncCompletion = getSyncCallback(sync); + + appendRollbackRecord(txID, sync, syncCompletion); + + if (syncCompletion != null) + { + syncCompletion.waitCompletion(); + } + + } + + public void appendDeleteRecord(final long id, final boolean sync) throws Exception + { + SyncIOCompletion callback = getSyncCallback(sync); + + appendDeleteRecord(id, sync, callback); + + if (callback != null) + { + callback.waitCompletion(); + } + } + + protected SyncIOCompletion getSyncCallback(final boolean sync) + { + if (hasCallbackSupport) + { + if (sync) + { + return new SimpleWaitIOCallback(); + } + return DummyCallback.getInstance(); + } + return null; + } + + private static class NullEncoding implements EncodingSupport + { + + private static NullEncoding instance = new NullEncoding(); + + public void decode(final HornetQBuffer buffer) + { + // no-op + } + + public void encode(final HornetQBuffer buffer) + { + // no-op + } + + public int getEncodeSize() + { + return 0; + } + } + +} Modified: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java =================================================================== --- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-08-09 09:43:05 UTC (rev 11156) +++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-08-09 10:26:46 UTC (rev 11157) @@ -72,7 +72,7 @@ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> */ -public class JournalImpl implements TestableJournal, JournalRecordProvider +public class JournalImpl extends JournalBase implements TestableJournal, JournalRecordProvider { private enum JournalState @@ -242,10 +242,7 @@ final int maxAIO, final int userVersion) { - if (fileFactory == null) - { - throw new NullPointerException("fileFactory is null"); - } + super(fileFactory.isSupportsCallbacks()); if (fileSize < JournalImpl.MIN_FILE_SIZE) { throw new IllegalArgumentException("File size cannot be less than " + JournalImpl.MIN_FILE_SIZE + " bytes"); @@ -793,34 +790,9 @@ // Journal implementation // ---------------------------------------------------------------- - public void appendAddRecord(final long id, final byte recordType, final byte[] record, final boolean sync) throws Exception - { - appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync); - } - + @Override public void appendAddRecord(final long id, final byte recordType, - final byte[] record, - final boolean sync, - final IOCompletion callback) throws Exception - { - appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync, callback); - } - - public void appendAddRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync) throws Exception - { - SyncIOCompletion callback = getSyncCallback(sync); - - appendAddRecord(id, recordType, record, sync, callback); - - if (callback != null) - { - callback.waitCompletion(); - } - } - - public void appendAddRecord(final long id, - final byte recordType, final EncodingSupport record, final boolean sync, final IOCompletion callback) throws Exception @@ -865,34 +837,9 @@ } } - public void appendUpdateRecord(final long id, final byte recordType, final byte[] record, final boolean sync) throws Exception - { - appendUpdateRecord(id, recordType, new ByteArrayEncoding(record), sync); - } - + @Override public void appendUpdateRecord(final long id, final byte recordType, - final byte[] record, - final boolean sync, - final IOCompletion callback) throws Exception - { - appendUpdateRecord(id, recordType, new ByteArrayEncoding(record), sync, callback); - } - - public void appendUpdateRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync) throws Exception - { - SyncIOCompletion callback = getSyncCallback(sync); - - appendUpdateRecord(id, recordType, record, sync, callback); - - if (callback != null) - { - callback.waitCompletion(); - } - } - - public void appendUpdateRecord(final long id, - final byte recordType, final EncodingSupport record, final boolean sync, final IOCompletion callback) throws Exception @@ -956,18 +903,8 @@ } } - public void appendDeleteRecord(final long id, final boolean sync) throws Exception - { - SyncIOCompletion callback = getSyncCallback(sync); - appendDeleteRecord(id, sync, callback); - - if (callback != null) - { - callback.waitCompletion(); - } - } - + @Override public void appendDeleteRecord(final long id, final boolean sync, final IOCompletion callback) throws Exception { checkJournalIsLoaded(); @@ -1035,12 +972,7 @@ } } - public void appendAddRecordTransactional(final long txID, final long id, final byte recordType, final byte[] record) throws Exception - { - appendAddRecordTransactional(txID, id, recordType, new ByteArrayEncoding(record)); - - } - + @Override public void appendAddRecordTransactional(final long txID, final long id, final byte recordType, @@ -1099,17 +1031,10 @@ state = newState; } + @Override public void appendUpdateRecordTransactional(final long txID, final long id, final byte recordType, - final byte[] record) throws Exception - { - appendUpdateRecordTransactional(txID, id, recordType, new ByteArrayEncoding(record)); - } - - public void appendUpdateRecordTransactional(final long txID, - final long id, - final byte recordType, final EncodingSupport record) throws Exception { checkJournalIsLoaded(); @@ -1151,11 +1076,8 @@ } } - public void appendDeleteRecordTransactional(final long txID, final long id, final byte[] record) throws Exception - { - appendDeleteRecordTransactional(txID, id, new ByteArrayEncoding(record)); - } + @Override public void appendDeleteRecordTransactional(final long txID, final long id, final EncodingSupport record) throws Exception { checkJournalIsLoaded(); @@ -1195,39 +1117,6 @@ } } - public void appendDeleteRecordTransactional(final long txID, final long id) throws Exception - { - appendDeleteRecordTransactional(txID, id, NullEncoding.instance); - } - - public void appendPrepareRecord(final long txID, - final byte[] transactionData, - final boolean sync, - final IOCompletion completion) throws Exception - { - appendPrepareRecord(txID, new ByteArrayEncoding(transactionData), sync, completion); - } - - /* (non-Javadoc) - * @see org.hornetq.core.journal.Journal#appendPrepareRecord(long, byte[], boolean) - */ - public void appendPrepareRecord(final long txID, final byte[] transactionData, final boolean sync) throws Exception - { - appendPrepareRecord(txID, new ByteArrayEncoding(transactionData), sync); - } - - public void appendPrepareRecord(final long txID, final EncodingSupport transactionData, final boolean sync) throws Exception - { - SyncIOCompletion syncCompletion = getSyncCallback(sync); - - appendPrepareRecord(txID, transactionData, sync, syncCompletion); - - if (syncCompletion != null) - { - syncCompletion.waitCompletion(); - } - } - /** * * <p>If the system crashed after a prepare was called, it should store information that is required to bring the transaction @@ -1241,6 +1130,7 @@ * @param transactionData extra user data for the prepare * @throws Exception */ + @Override public void appendPrepareRecord(final long txID, final EncodingSupport transactionData, final boolean sync, @@ -1287,18 +1177,6 @@ } } - public void appendCommitRecord(final long txID, final boolean sync) throws Exception - { - SyncIOCompletion syncCompletion = getSyncCallback(sync); - - appendCommitRecord(txID, sync, syncCompletion, true); - - if (syncCompletion != null) - { - syncCompletion.waitCompletion(); - } - } - /* (non-Javadoc) * @see org.hornetq.core.journal.Journal#lineUpContex(org.hornetq.core.journal.IOCompletion) */ @@ -1307,12 +1185,7 @@ callback.storeLineUp(); } - public void appendCommitRecord(final long txID, final boolean sync, final IOCompletion callback) throws Exception - { - appendCommitRecord(txID, sync, callback, true); - } - /** * <p>A transaction record (Commit or Prepare), will hold the number of elements the transaction has on each file.</p> * <p>For example, a transaction was spread along 3 journal files with 10 pendingTransactions on each file. @@ -1327,9 +1200,8 @@ * <p> If for any reason there are missing pendingTransactions, that means the transaction was not completed and we should ignore the whole transaction </p> * <p> We can't just use a global counter as reclaiming could delete files after the transaction was successfully committed. * That also means not having a whole file on journal-reload doesn't mean we have to invalidate the transaction </p> - * */ - + @Override public void appendCommitRecord(final long txID, final boolean sync, final IOCompletion callback, boolean lineUpContext) throws Exception { checkJournalIsLoaded(); @@ -1376,19 +1248,7 @@ } } - public void appendRollbackRecord(final long txID, final boolean sync) throws Exception - { - SyncIOCompletion syncCompletion = getSyncCallback(sync); - - appendRollbackRecord(txID, sync, syncCompletion); - - if (syncCompletion != null) - { - syncCompletion.waitCompletion(); - } - - } - + @Override public void appendRollbackRecord(final long txID, final boolean sync, final IOCompletion callback) throws Exception { checkJournalIsLoaded(); @@ -1475,6 +1335,7 @@ { return load(committedRecords, preparedTransactions, failureCallback, true); } + /** * @see JournalImpl#load(LoaderCallback) */ @@ -3058,25 +2919,6 @@ return tx; } - private SyncIOCompletion getSyncCallback(final boolean sync) - { - if (fileFactory.isSupportsCallbacks()) - { - if (sync) - { - return new SimpleWaitIOCallback(); - } - else - { - return DummyCallback.getInstance(); - } - } - else - { - return null; - } - } - /** * @return * @throws Exception @@ -3171,26 +3013,6 @@ // Inner classes // --------------------------------------------------------------------------- - private static class NullEncoding implements EncodingSupport - { - - private static NullEncoding instance = new NullEncoding(); - - public void decode(final HornetQBuffer buffer) - { - } - - public void encode(final HornetQBuffer buffer) - { - } - - public int getEncodeSize() - { - return 0; - } - - } - // Used on Load private static class TransactionHolder { Modified: 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-09 09:43:05 UTC (rev 11156) +++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java 2011-08-09 10:26:46 UTC (rev 11157) @@ -14,7 +14,6 @@ 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.ByteArrayEncoding; import org.hornetq.core.journal.impl.dataformat.JournalAddRecord; import org.hornetq.core.journal.impl.dataformat.JournalInternalRecord; @@ -22,23 +21,22 @@ * 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 like a Journal would but without verifying records. + * Its main purpose is to store the data as a Journal would, but without verifying records. */ -public class ReplicatingJournal implements Journal +public class ReplicatingJournal extends JournalBase implements Journal { private final ReentrantLock lockAppend = new ReentrantLock(); private final ReadWriteLock journalLock = new ReentrantReadWriteLock(); - private final JournalFile file; - private final boolean hasCallbackSupport; + private final JournalFile currentFile; /** * @param file */ public ReplicatingJournal(JournalFile file, boolean hasCallbackSupport) { - this.file = file; - this.hasCallbackSupport = hasCallbackSupport; + super(hasCallbackSupport); + this.currentFile = file; } @Override @@ -60,32 +58,7 @@ } // ------------------------ - @Override - public void appendAddRecord(long id, byte recordType, byte[] record, boolean sync) throws Exception - { - appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync); - } - @Override - public void appendAddRecord(long id, byte recordType, byte[] record, boolean sync, IOCompletion completionCallback) - throws Exception - { - appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync, completionCallback); - } - - @Override - public void appendAddRecord(long id, byte recordType, EncodingSupport record, boolean sync) throws Exception - { - SyncIOCompletion callback = getSyncCallback(sync); - - appendAddRecord(id, recordType, record, sync, callback); - - if (callback != null) - { - callback.waitCompletion(); - } - } - // ------------------------ private void readLockJournal() @@ -99,84 +72,52 @@ } @Override - public void appendAddRecord(long id, byte recordType, EncodingSupport record, boolean sync, - IOCompletion callback) throws Exception + public void appendAddRecord(long id, byte recordType, EncodingSupport record, boolean sync, IOCompletion callback) + throws Exception { JournalInternalRecord addRecord = new JournalAddRecord(true, id, recordType, record); - if (callback != null) - { - callback.storeLineUp(); - } + if (callback != null) + { + callback.storeLineUp(); + } - lockAppend.lock(); - try - { - JournalFile usedFile = appendRecord(addRecord, false, sync, null, callback); - } - finally - { - lockAppend.unlock(); - } - + lockAppend.lock(); + try + { + appendRecord(addRecord, sync, callback); + } + finally + { + lockAppend.unlock(); + } } /** - * @param addRecord - * @param b - * @param sync - * @param object - * @param callback - * @return + * Write the record to the current file. */ - private JournalFile appendRecord(JournalInternalRecord addRecord, boolean b, boolean sync, Object object, - IOCompletion callback) + private void appendRecord(JournalInternalRecord encoder, boolean sync, IOCompletion callback) throws Exception { - // TODO Auto-generated method stub - return null; - } + encoder.setFileID(currentFile.getRecordID()); - @Override - public void appendUpdateRecord(long id, byte recordType, byte[] record, boolean sync) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); + if (callback != null) + { + currentFile.getFile().write(encoder, sync, callback); + } + else + { + currentFile.getFile().write(encoder, sync); + } } @Override - public void - appendUpdateRecord(long id, byte recordType, byte[] record, boolean sync, IOCompletion completionCallback) - throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override - public void appendUpdateRecord(long id, byte recordType, EncodingSupport record, boolean sync) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override - public void appendUpdateRecord(long id, byte recordType, EncodingSupport record, boolean sync, - IOCompletion completionCallback) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override - public void appendDeleteRecord(long id, boolean sync) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override public void appendDeleteRecord(long id, boolean sync, IOCompletion completionCallback) throws Exception { throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); } @Override - public void appendAddRecordTransactional(long txID, long id, byte recordType, byte[] record) throws Exception + public void appendDeleteRecordTransactional(long txID, long id, EncodingSupport record) throws Exception { throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); } @@ -189,7 +130,9 @@ } @Override - public void appendUpdateRecordTransactional(long txID, long id, byte recordType, byte[] record) throws Exception + public void + appendUpdateRecord(long id, byte recordType, EncodingSupport record, boolean sync, IOCompletion callback) + throws Exception { throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); } @@ -202,36 +145,6 @@ } @Override - public void appendDeleteRecordTransactional(long txID, long id, byte[] record) 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 appendDeleteRecordTransactional(long txID, long id) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override - public void appendCommitRecord(long txID, boolean sync) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override - public void appendCommitRecord(long txID, boolean sync, IOCompletion callback) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override public void appendCommitRecord(long txID, boolean sync, IOCompletion callback, boolean lineUpContext) throws Exception { @@ -239,12 +152,6 @@ } @Override - public void appendPrepareRecord(long txID, EncodingSupport transactionData, boolean sync) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override public void appendPrepareRecord(long txID, EncodingSupport transactionData, boolean sync, IOCompletion callback) throws Exception { @@ -252,25 +159,6 @@ } @Override - public void appendPrepareRecord(long txID, byte[] transactionData, boolean sync) throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override - public void appendPrepareRecord(long txID, byte[] transactionData, boolean sync, IOCompletion callback) - throws Exception - { - throw new HornetQException(HornetQException.UNSUPPORTED_PACKET); - } - - @Override - public void appendRollbackRecord(long txID, boolean sync) 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); @@ -331,17 +219,4 @@ { throw new UnsupportedOperationException(); } - - private SyncIOCompletion getSyncCallback(final boolean sync) - { - if (hasCallbackSupport) - { - if (sync) - { - return new SimpleWaitIOCallback(); - } - return DummyCallback.getInstance(); - } - return null; - } }