[hornetq-commits] JBoss hornetq SVN: r8163 - in branches/Clebert_Sync: src/main/org/hornetq/core/server/impl and 1 other directories.

Author: clebert.suconic(a)jboss.com Date: 2009-10-28 23:22:06 -0400 (Wed, 28 Oct 2009) New Revision: 8163 Added: branches/Clebert_Sync/tests/src/org/hornetq/tests/integration/cluster/failover/JournalCopyFailoverTest.java Modified: branches/Clebert_Sync/src/main/org/hornetq/core/server/HornetQServer.java branches/Clebert_Sync/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java Log: Backup changes Modified: branches/Clebert_Sync/src/main/org/hornetq/core/server/HornetQServer.java =================================================================== --- branches/Clebert_Sync/src/main/org/hornetq/core/server/HornetQServer.java 2009-10-29 00:35:30 UTC (rev 8162) +++ branches/Clebert_Sync/src/main/org/hornetq/core/server/HornetQServer.java 2009-10-29 03:22:06 UTC (rev 8163) @@ -126,6 +126,8 @@ SimpleString filterString, boolean durable, boolean temporary) throws Exception; + + void configureBackup(String connectorClassName, String properties, boolean replicated) throws Exception; void destroyQueue(SimpleString queueName, ServerSession session) throws Exception; Modified: branches/Clebert_Sync/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Clebert_Sync/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-10-29 00:35:30 UTC (rev 8162) +++ branches/Clebert_Sync/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-10-29 03:22:06 UTC (rev 8163) @@ -100,6 +100,7 @@ import org.hornetq.core.transaction.ResourceManager; import org.hornetq.core.transaction.impl.ResourceManagerImpl; import org.hornetq.core.version.Version; +import org.hornetq.ra.Util; import org.hornetq.utils.ExecutorFactory; import org.hornetq.utils.HornetQThreadFactory; import org.hornetq.utils.OrderedExecutorFactory; @@ -506,6 +507,19 @@ return clusterManager; } + /** TODO: Update Client CFs + * This should also work with shared storage + */ + public void configureBackup(String connectorClassName, String config, boolean replicated) throws Exception + { + // TODO: The use of ra.Util is temporary, this class should be moved somewhere else after the merge. + TransportConfiguration transport = new TransportConfiguration(connectorClassName, Util.parseConfig(config)); + this.createReplication(transport); + this.replicationManager.start(); + storageManager.initiateReplication(replicationManager); + + } + public ReattachSessionResponseMessage reattachSession(final RemotingConnection connection, final String name, final int lastReceivedCommandID) throws Exception @@ -844,10 +858,7 @@ else { - replicationFailoverManager = createBackupConnection(backupConnector, threadPool, scheduledPool); - - this.replicationManager = new ReplicationManagerImpl(replicationFailoverManager, - this.executorFactory.getExecutor()); + createReplication(backupConnector); replicationManager.start(); } } @@ -855,7 +866,18 @@ return true; } + /** + * @param backupConnector + */ + private void createReplication(TransportConfiguration backupConnector) + { + replicationFailoverManager = createBackupConnection(backupConnector, threadPool, scheduledPool); + this.replicationManager = new ReplicationManagerImpl(replicationFailoverManager, + this.executorFactory.getExecutor()); + } + + private synchronized void callActivateCallbacks() { for (ActivateCallback callback : activateCallbacks) Added: branches/Clebert_Sync/tests/src/org/hornetq/tests/integration/cluster/failover/JournalCopyFailoverTest.java =================================================================== --- branches/Clebert_Sync/tests/src/org/hornetq/tests/integration/cluster/failover/JournalCopyFailoverTest.java (rev 0) +++ branches/Clebert_Sync/tests/src/org/hornetq/tests/integration/cluster/failover/JournalCopyFailoverTest.java 2009-10-29 03:22:06 UTC (rev 8163) @@ -0,0 +1,120 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.failover; + +import org.hornetq.core.buffers.ChannelBuffers; +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.client.ClientProducer; +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.utils.SimpleString; + +/** + * A JournalCopyFailoverTest + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class JournalCopyFailoverTest extends FailoverTestBase +{ + + // Constants ----------------------------------------------------- + + private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public void testSynchronousCopy() throws Exception + { + ClientSessionFactoryInternal sf = getSessionFactory(); + + sf.setBlockOnNonPersistentSend(true); + sf.setBlockOnPersistentSend(true); + + ClientSession session = sf.createSession(true, true); + + session.createQueue(ADDRESS, ADDRESS, true); + + ClientProducer prod = session.createProducer(ADDRESS); + + for (int i = 0; i < 100; i++) + { + ClientMessage msg = session.createClientMessage(true); + msg.setBody(ChannelBuffers.wrappedBuffer(new byte[512])); + prod.send(msg); + } + + server0Service.configureBackup("org.hornetq.integration.transports.netty.NettyConnectorFactory", + "hornetq.remoting.netty.port=" + (org.hornetq.integration.transports.netty.TransportConstants.DEFAULT_PORT + 1), + true); + + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + protected void createReplicatedConfigs() throws Exception + { + Configuration config1 = super.createDefaultConfig(); + config1.setBindingsDirectory(config1.getBindingsDirectory() + "_backup"); + config1.setJournalDirectory(config1.getJournalDirectory() + "_backup"); + config1.setPagingDirectory(config1.getPagingDirectory() + "_backup"); + config1.setLargeMessagesDirectory(config1.getLargeMessagesDirectory() + "_backup"); + config1.getAcceptorConfigurations().clear(); + config1.getAcceptorConfigurations().add(getAcceptorTransportConfiguration(false)); + config1.setSecurityEnabled(false); + config1.setSharedStore(false); + config1.setBackup(true); + server1Service = super.createServer(true, config1); + + Configuration config0 = super.createDefaultConfig(); + config0.getAcceptorConfigurations().clear(); + config0.getAcceptorConfigurations().add(getAcceptorTransportConfiguration(true)); + getConnectorTransportConfiguration(false); + // config0.getConnectorConfigurations().put("toBackup", getConnectorTransportConfiguration(false)); + // config0.setBackupConnectorName("toBackup"); + // config0.setSecurityEnabled(false); + // config0.setSharedStore(false); + server0Service = super.createServer(true, config0); + + server1Service.start(); + server0Service.start(); + } + + @Override + protected TransportConfiguration getAcceptorTransportConfiguration(boolean live) + { + return getNettyAcceptorTransportConfiguration(live); + } + + @Override + protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) + { + return getNettyConnectorTransportConfiguration(live); + } + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +}