[hornetq-commits] JBoss hornetq SVN: r8006 - in branches/Replication_Clebert: src/main/org/hornetq/core/persistence and 6 other directories.

Author: clebert.suconic(a)jboss.com Date: 2009-09-28 21:14:33 -0400 (Mon, 28 Sep 2009) New Revision: 8006 Modified: branches/Replication_Clebert/src/main/org/hornetq/core/journal/Journal.java branches/Replication_Clebert/src/main/org/hornetq/core/journal/TestableJournal.java branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Replication_Clebert/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java Log: Changes Modified: branches/Replication_Clebert/src/main/org/hornetq/core/journal/Journal.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/journal/Journal.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/journal/Journal.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -74,7 +74,13 @@ void appendRollbackRecord(long txID, boolean sync) throws Exception; // Load + + /** This method could be promoted to {@link Journal} interface when we decide to use the loadManager + * instead of load(List,List) + */ + long load(LoaderCallback reloadManager) throws Exception; + long load(List<RecordInfo> committedRecords, List<PreparedTransactionInfo> preparedTransactions, TransactionFailureCallback transactionFailure) throws Exception; int getAlignment() throws Exception; Modified: branches/Replication_Clebert/src/main/org/hornetq/core/journal/TestableJournal.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/journal/TestableJournal.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/journal/TestableJournal.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -46,11 +46,6 @@ int getMaxAIO(); - /** This method could be promoted to {@link Journal} interface when we decide to use the loadManager - * instead of load(List,List) - */ - long load(LoaderCallback reloadManager) throws Exception; - void forceMoveNextFile() throws Exception; void setAutoReclaim(boolean autoReclaim); Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -96,6 +96,10 @@ void deletePageTransactional(long txID, long recordID) throws Exception; + /** This method is only useful at the backup side. We only load internal structures making the journals ready for + * append mode on the backup side. */ + void loadInternalOnly() throws Exception; + void loadMessageJournal(PagingManager pagingManager, ResourceManager resourceManager, Map<Long, Queue> queues, Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -35,6 +35,7 @@ import org.hornetq.core.filter.Filter; import org.hornetq.core.journal.EncodingSupport; import org.hornetq.core.journal.Journal; +import org.hornetq.core.journal.LoaderCallback; import org.hornetq.core.journal.PreparedTransactionInfo; import org.hornetq.core.journal.RecordInfo; import org.hornetq.core.journal.SequentialFile; @@ -1087,7 +1088,42 @@ { return started; } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#loadInternalOnly() + */ + public void loadInternalOnly() throws Exception + { + LoaderCallback dummyLoader = new LoaderCallback() + { + + public void failedTransaction(long transactionID, List<RecordInfo> records, List<RecordInfo> recordsToDelete) + { + } + + public void updateRecord(RecordInfo info) + { + } + + public void deleteRecord(long id) + { + } + + public void addRecord(RecordInfo info) + { + } + + public void addPreparedTransaction(PreparedTransactionInfo preparedTransaction) + { + } + }; + + bindingsJournal.load(dummyLoader); + messageJournal.load(dummyLoader); + } + + // Public ----------------------------------------------------------------------------------- public Journal getMessageJournal() Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -238,4 +238,11 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#loadInternalOnly() + */ + public void loadInternalOnly() throws Exception + { + } + } Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -14,6 +14,7 @@ package org.hornetq.core.replication; import org.hornetq.core.remoting.ChannelHandler; +import org.hornetq.core.server.HornetQComponent; /** * A ReplicationEndpoint @@ -22,7 +23,7 @@ * * */ -public interface ReplicationEndpoint extends ChannelHandler +public interface ReplicationEndpoint extends ChannelHandler, HornetQComponent { } Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -33,7 +33,7 @@ // Constants ----------------------------------------------------- // Attributes ---------------------------------------------------- - + private final HornetQServer server; // Static -------------------------------------------------------- @@ -41,7 +41,7 @@ // Constructors -------------------------------------------------- public ReplicationEndpointImpl(HornetQServer server) { - this.server = server ; + this.server = server; } // Public -------------------------------------------------------- @@ -53,6 +53,28 @@ } + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQComponent#isStarted() + */ + public boolean isStarted() + { + return true; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQComponent#start() + */ + public void start() throws Exception + { + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQComponent#stop() + */ + public void stop() throws Exception + { + } + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Modified: branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -199,6 +199,8 @@ private ConnectionManager replicatingConnectionManager; private ReplicationManager replicationManager; + + private ReplicationEndpoint replicationEndpoint = new ReplicationEndpointImpl(this); private final Set<ActivateCallback> activateCallbacks = new HashSet<ActivateCallback>(); @@ -593,7 +595,12 @@ public synchronized ReplicationEndpoint createReplicationEndpoint() { - return new ReplicationEndpointImpl(this); + if (replicationEndpoint == null) + { + replicationEndpoint = new ReplicationEndpointImpl(this); + + } + return replicationEndpoint; } public void removeSession(final String name) throws Exception @@ -891,21 +898,24 @@ { // Handle backup server activation - if (configuration.isSharedStore()) + if (!configuration.isSharedStore()) { - // Complete the startup procedure + if (replicationEndpoint == null) + { + log.warn("There is no replication endpoint, can't activate this backup server"); + throw new HornetQException(HornetQException.INTERNAL_ERROR, "Can't activate the server"); + } + + replicationEndpoint.stop(); + } + + // Complete the startup procedure - log.info("Activating server"); + log.info("Activating server"); - configuration.setBackup(false); + configuration.setBackup(false); - initialisePart2(); - } - else - { - // TODO - // just load journal - } + initialisePart2(); } return true; Modified: branches/Replication_Clebert/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- branches/Replication_Clebert/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-09-28 22:26:08 UTC (rev 8005) +++ branches/Replication_Clebert/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-09-29 01:14:33 UTC (rev 8006) @@ -1095,6 +1095,13 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#loadInternalOnly() + */ + public void loadInternalOnly() throws Exception + { + } + } class FakeStoreFactory implements PagingStoreFactory