[hornetq-commits] JBoss hornetq SVN: r8988 - in trunk: src/main/org/hornetq/core/server and 7 other directories.

Author: clebert.suconic(a)jboss.com Date: 2010-03-28 22:53:39 -0400 (Sun, 28 Mar 2010) New Revision: 8988 Added: trunk/tests/src/org/hornetq/tests/integration/jms/cluster/ReplicatedJMSFailoverTest.java Modified: trunk/src/main/org/hornetq/core/replication/ReplicationEndpoint.java trunk/src/main/org/hornetq/core/server/HornetQServer.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/src/main/org/hornetq/jms/persistence/JMSStorageManager.java trunk/src/main/org/hornetq/jms/persistence/impl/journal/JournalJMSStorageManagerImpl.java trunk/src/main/org/hornetq/jms/persistence/impl/nullpm/NullJMSStorageManagerImpl.java trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java Log: Replication on JMS journal Modified: trunk/src/main/org/hornetq/core/replication/ReplicationEndpoint.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/ReplicationEndpoint.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/src/main/org/hornetq/core/replication/ReplicationEndpoint.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -14,6 +14,7 @@ package org.hornetq.core.replication; import org.hornetq.api.core.HornetQException; +import org.hornetq.core.journal.Journal; import org.hornetq.core.journal.JournalLoadInformation; import org.hornetq.core.protocol.core.Channel; import org.hornetq.core.protocol.core.ChannelHandler; @@ -35,4 +36,6 @@ void compareJournalInformation(JournalLoadInformation[] journalInformation) throws HornetQException; + void registerJournal(final byte id, final Journal journal); + } Modified: trunk/src/main/org/hornetq/core/server/HornetQServer.java =================================================================== --- trunk/src/main/org/hornetq/core/server/HornetQServer.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/src/main/org/hornetq/core/server/HornetQServer.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -26,6 +26,7 @@ import org.hornetq.core.protocol.core.Channel; import org.hornetq.core.remoting.server.RemotingService; import org.hornetq.core.replication.ReplicationEndpoint; +import org.hornetq.core.replication.ReplicationManager; import org.hornetq.core.security.Role; import org.hornetq.core.server.cluster.ClusterManager; import org.hornetq.core.server.group.GroupingHandler; @@ -135,6 +136,10 @@ void setGroupingHandler(GroupingHandler groupingHandler); GroupingHandler getGroupingHandler(); + + ReplicationEndpoint getReplicationEndpoint(); + + ReplicationManager getReplicationManager(); boolean checkActivate() throws Exception; } Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -750,15 +750,19 @@ return groupingHandler; } - // Public - // --------------------------------------------------------------------------------------- - - /** For tests only */ public ReplicationEndpoint getReplicationEndpoint() { return replicationEndpoint; } + + public ReplicationManager getReplicationManager() + { + return replicationManager; + } + // Public + // --------------------------------------------------------------------------------------- + // Package protected // ---------------------------------------------------------------------------- Modified: trunk/src/main/org/hornetq/jms/persistence/JMSStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/jms/persistence/JMSStorageManager.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/src/main/org/hornetq/jms/persistence/JMSStorageManager.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -15,6 +15,7 @@ import java.util.List; +import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.core.server.HornetQComponent; import org.hornetq.jms.persistence.config.PersistedConnectionFactory; import org.hornetq.jms.persistence.config.PersistedDestination; @@ -40,7 +41,9 @@ // Constructors -------------------------------------------------- // Public -------------------------------------------------------- - + + void load() throws Exception; + void storeDestination(PersistedDestination destination) throws Exception; void deleteDestination(PersistedType type, String name) throws Exception; @@ -68,4 +71,11 @@ void deleteJNDI(PersistedType type, String name, String address) throws Exception; void deleteJNDI(PersistedType type, String name) throws Exception; + + /** + * Add the journal here to the replication endpoint + * @param replicationEndpoint + * @throws Exception + */ + void installReplication(ReplicationEndpoint replicationEndpoint) throws Exception; } Modified: trunk/src/main/org/hornetq/jms/persistence/impl/journal/JournalJMSStorageManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/persistence/impl/journal/JournalJMSStorageManagerImpl.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/src/main/org/hornetq/jms/persistence/impl/journal/JournalJMSStorageManagerImpl.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -22,7 +22,6 @@ import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQBuffers; import org.hornetq.api.core.Pair; -import org.hornetq.api.core.SimpleString; import org.hornetq.core.config.Configuration; import org.hornetq.core.journal.Journal; import org.hornetq.core.journal.PreparedTransactionInfo; @@ -30,6 +29,7 @@ import org.hornetq.core.journal.SequentialFileFactory; import org.hornetq.core.journal.impl.JournalImpl; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; +import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.core.replication.ReplicationManager; import org.hornetq.core.replication.impl.ReplicatedJournal; import org.hornetq.core.server.JournalType; @@ -38,7 +38,6 @@ import org.hornetq.jms.persistence.config.PersistedDestination; import org.hornetq.jms.persistence.config.PersistedJNDI; import org.hornetq.jms.persistence.config.PersistedType; -import org.hornetq.jms.server.config.ConnectionFactoryConfiguration; import org.hornetq.utils.IDGenerator; /** @@ -290,27 +289,31 @@ jmsJournal.start(); - load(); - started = true; } + /* (non-Javadoc) + * @see org.hornetq.jms.persistence.JMSStorageManager#installReplication(org.hornetq.core.replication.ReplicationEndpoint) + */ + public void installReplication(ReplicationEndpoint replicationEndpoint) throws Exception + { + jmsJournal.loadInternalOnly(); + replicationEndpoint.registerJournal((byte)2, this.jmsJournal); + } + + + /* (non-Javadoc) * @see org.hornetq.core.server.HornetQComponent#stop() */ public void stop() throws Exception { + this.started = false; jmsJournal.stop(); } - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - private void load() throws Exception + public void load() throws Exception { mapFactories.clear(); @@ -359,6 +362,13 @@ } + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + private void checkAndCreateDir(final String dir, final boolean create) { File f = new File(dir); @@ -380,6 +390,7 @@ } + // Inner classes ------------------------------------------------- } Modified: trunk/src/main/org/hornetq/jms/persistence/impl/nullpm/NullJMSStorageManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/persistence/impl/nullpm/NullJMSStorageManagerImpl.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/src/main/org/hornetq/jms/persistence/impl/nullpm/NullJMSStorageManagerImpl.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -16,6 +16,7 @@ import java.util.Collections; import java.util.List; +import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.jms.persistence.JMSStorageManager; import org.hornetq.jms.persistence.config.PersistedConnectionFactory; import org.hornetq.jms.persistence.config.PersistedDestination; @@ -136,6 +137,20 @@ return Collections.emptyList(); } + /* (non-Javadoc) + * @see org.hornetq.jms.persistence.JMSStorageManager#installReplication(org.hornetq.core.replication.ReplicationEndpoint) + */ + public void installReplication(ReplicationEndpoint replicationEndpoint) throws Exception + { + } + + /* (non-Javadoc) + * @see org.hornetq.jms.persistence.JMSStorageManager#load() + */ + public void load() throws Exception + { + } + // Constants ----------------------------------------------------- // Attributes ---------------------------------------------------- Modified: trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -13,7 +13,14 @@ package org.hornetq.jms.server.impl; -import java.util.*; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; import javax.naming.Context; import javax.naming.InitialContext; @@ -33,7 +40,6 @@ import org.hornetq.core.deployers.impl.FileDeploymentManager; import org.hornetq.core.deployers.impl.XmlDeployer; import org.hornetq.core.logging.Logger; -import org.hornetq.core.persistence.impl.journal.JournalStorageManager; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.BindingType; import org.hornetq.core.security.Role; @@ -125,12 +131,16 @@ { this.server = server; + this.coreConfig = server.getConfiguration(); + configFileName = null; } public JMSServerManagerImpl(final HornetQServer server, final String configFileName) throws Exception { this.server = server; + + this.coreConfig = server.getConfiguration(); this.configFileName = configFileName; } @@ -138,6 +148,8 @@ public JMSServerManagerImpl(final HornetQServer server, final JMSConfiguration configuration) throws Exception { this.server = server; + + this.coreConfig = server.getConfiguration(); configFileName = null; @@ -213,9 +225,16 @@ deploymentManager = new FileDeploymentManager(server.getConfiguration().getFileDeployerScanPeriod()); server.registerActivateCallback(this); - + server.start(); + + if (server.getReplicationEndpoint() != null) + { + createJournal(); + storage.installReplication(server.getReplicationEndpoint()); + } + started = true; } @@ -235,11 +254,18 @@ { deploymentManager.stop(); } + + // Storage could be null on a shared store backup server before initialization + if (storage != null) + { + storage.stop(); + } + + unbindJNDI(queueJNDI); - // for (String destination : destinationBindings.keySet()) - // { - // undeployDestination(destination); - // } + unbindJNDI(topicJNDI); + + unbindJNDI(connectionFactoryJNDI); for (String connectionFactory : new HashSet<String>(connectionFactories.keySet())) { @@ -255,10 +281,14 @@ context.close(); } - jmsManagementService.unregisterJMSServer(); - - jmsManagementService.stop(); - + // it could be null if a backup + if (jmsManagementService != null) + { + jmsManagementService.unregisterJMSServer(); + + jmsManagementService.stop(); + } + server.stop(); started = false; @@ -1338,27 +1368,38 @@ } /** - * @param server + * @param param */ - private void initJournal() throws Exception + private void unbindJNDI(Map<String, List<String>> param) { - this.coreConfig = server.getConfiguration(); - - if (storage == null) + for (List<String> elementList : param.values()) { - if (coreConfig.isPersistenceEnabled()) + for (String key : elementList) { - // TODO: replication - storage = new JournalJMSStorageManagerImpl(new TimeAndCounterIDGenerator(), coreConfig, null); + try + { + context.unbind(key); + } + catch (Exception e) + { + log.warn("Impossible to unbind key " + key + " from JNDI"); + } } - else - { - storage = new NullJMSStorageManagerImpl(); - } } + } - storage.start(); + /** + * @param server + */ + private void initJournal() throws Exception + { + this.coreConfig = server.getConfiguration(); + + createJournal(); + + storage.load(); + List<PersistedConnectionFactory> cfs = storage.recoverConnectionFactories(); for (PersistedConnectionFactory cf : cfs) @@ -1428,6 +1469,33 @@ } } + /** + * @throws Exception + */ + private void createJournal() throws Exception + { + if (storage == null) + { + if (coreConfig.isPersistenceEnabled()) + { + storage = new JournalJMSStorageManagerImpl(new TimeAndCounterIDGenerator(), server.getConfiguration(), server.getReplicationManager()); + } + else + { + storage = new NullJMSStorageManagerImpl(); + } + } + else + { + if (storage.isStarted()) + { + storage.stop(); + } + } + + storage.start(); + } + private synchronized boolean removeFromJNDI(final Map<String, List<String>> jndiMap, final String name) throws Exception { checkInitialised(); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -26,6 +26,7 @@ import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; +import javax.naming.NamingException; import junit.framework.Assert; @@ -45,7 +46,10 @@ import org.hornetq.jms.client.HornetQConnectionFactory; import org.hornetq.jms.client.HornetQDestination; import org.hornetq.jms.client.HornetQSession; +import org.hornetq.jms.server.JMSServerManager; +import org.hornetq.jms.server.impl.JMSServerManagerImpl; import org.hornetq.spi.core.protocol.RemotingConnection; +import org.hornetq.tests.unit.util.InVMContext; import org.hornetq.tests.util.RandomUtil; import org.hornetq.tests.util.UnitTestCase; @@ -69,12 +73,24 @@ // Constants ----------------------------------------------------- // Attributes ---------------------------------------------------- + + protected InVMContext ctx1 = new InVMContext(); + + protected InVMContext ctx2 = new InVMContext(); + + protected Configuration backupConf; + + protected Configuration liveConf; + + protected JMSServerManager liveJMSService; - private HornetQServer liveService; + protected HornetQServer liveService; + + protected JMSServerManager backupJMSService; - private HornetQServer backupService; + protected HornetQServer backupService; - private Map<String, Object> backupParams = new HashMap<String, Object>(); + protected Map<String, Object> backupParams = new HashMap<String, Object>(); // Static -------------------------------------------------------- @@ -82,6 +98,63 @@ // Public -------------------------------------------------------- + public void testCreateQueue() throws Exception + { + liveJMSService.createQueue("queue1", null, true, "/queue/queue1"); + assertNotNull(ctx1.lookup("/queue/queue1")); + liveJMSService.stop(); + + Object obj = null; + + try + { + obj = ctx1.lookup("/queue/queue1"); + } + catch (NamingException expected) + { + + } + + assertNull(obj); + + backupJMSService.stop(); + + backupConf.setBackup(false); + + backupJMSService.start(); + + assertNotNull(ctx2.lookup("/queue/queue1")); + } + + + public void testCreateTopic() throws Exception + { + liveJMSService.createTopic("topic", "/topic/t1"); + assertNotNull(ctx1.lookup("//topic/t1")); + liveJMSService.stop(); + + Object obj = null; + + try + { + obj = ctx1.lookup("//topic/t1"); + } + catch (NamingException expected) + { + + } + + assertNull(obj); + + backupJMSService.stop(); + + backupConf.setBackup(false); + + backupJMSService.start(); + + assertNotNull(ctx2.lookup("/topic/t1")); + } + public void testAutomaticFailover() throws Exception { HornetQConnectionFactory jbcf = (HornetQConnectionFactory) HornetQJMSClient.createConnectionFactory(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), @@ -258,7 +331,15 @@ { super.setUp(); - Configuration backupConf = new ConfigurationImpl(); + startServers(); + } + + /** + * @throws Exception + */ + protected void startServers() throws Exception + { + backupConf = new ConfigurationImpl(); backupConf.setSecurityEnabled(false); backupParams.put(TransportConstants.SERVER_ID_PROP_NAME, 1); backupConf.getAcceptorConfigurations() @@ -271,9 +352,16 @@ backupConf.setPagingDirectory(getPageDir()); backupConf.setLargeMessagesDirectory(getLargeMessagesDir()); backupService = HornetQServers.newHornetQServer(backupConf, true); - backupService.start(); - Configuration liveConf = new ConfigurationImpl(); + backupJMSService = new JMSServerManagerImpl(backupService); + + backupJMSService.setContext(ctx2); + + backupJMSService.start(); + + + + liveConf = new ConfigurationImpl(); liveConf.setSecurityEnabled(false); liveConf.getAcceptorConfigurations() .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); @@ -285,19 +373,33 @@ liveConf.setLargeMessagesDirectory(getLargeMessagesDir()); liveService = HornetQServers.newHornetQServer(liveConf, true); - liveService.start(); + + liveJMSService = new JMSServerManagerImpl(liveService); + + liveJMSService.setContext(ctx1); + + liveJMSService.start(); + } @Override protected void tearDown() throws Exception { - backupService.stop(); + backupJMSService.stop(); - liveService.stop(); + liveJMSService.stop(); Assert.assertEquals(0, InVMRegistry.instance.size()); liveService = null; + + liveJMSService = null; + + backupJMSService = null; + + ctx1 = null; + + ctx2 = null; backupService = null; Added: trunk/tests/src/org/hornetq/tests/integration/jms/cluster/ReplicatedJMSFailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/cluster/ReplicatedJMSFailoverTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/jms/cluster/ReplicatedJMSFailoverTest.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -0,0 +1,102 @@ +/* + * Copyright 2010 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.jms.cluster; + +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.remoting.impl.invm.TransportConstants; +import org.hornetq.core.server.HornetQServers; +import org.hornetq.jms.server.impl.JMSServerManagerImpl; + +/** + * A ReplicatedJMSFailoverTest + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class ReplicatedJMSFailoverTest extends JMSFailoverTest +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + /** + * @throws Exception + */ + @Override + protected void startServers() throws Exception + { + backupConf = new ConfigurationImpl(); + backupConf.setSecurityEnabled(false); + backupParams.put(TransportConstants.SERVER_ID_PROP_NAME, 1); + backupConf.getAcceptorConfigurations() + .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory", backupParams)); + backupConf.setBackup(true); + backupConf.setSharedStore(false); + backupConf.setBindingsDirectory(getBindingsDir(0, true)); + backupConf.setJournalMinFiles(2); + backupConf.setJournalDirectory(getJournalDir(0, true)); + backupConf.setPagingDirectory(getPageDir(0, true)); + backupConf.setLargeMessagesDirectory(getLargeMessagesDir(0, true)); + backupService = HornetQServers.newHornetQServer(backupConf, true); + + backupJMSService = new JMSServerManagerImpl(backupService); + + backupJMSService.setContext(ctx2); + + backupJMSService.start(); + + + + liveConf = new ConfigurationImpl(); + liveConf.setSecurityEnabled(false); + + liveConf.getConnectorConfigurations().put("toBackup", new TransportConfiguration(INVM_CONNECTOR_FACTORY, backupParams)); + liveConf.setBackupConnectorName("toBackup"); + + liveConf.getAcceptorConfigurations() + .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); + liveConf.setSharedStore(false); + liveConf.setBindingsDirectory(getBindingsDir(0, false)); + liveConf.setJournalMinFiles(2); + liveConf.setJournalDirectory(getJournalDir(0, false)); + liveConf.setPagingDirectory(getPageDir(0, false)); + liveConf.setLargeMessagesDirectory(getLargeMessagesDir(0, false)); + + liveService = HornetQServers.newHornetQServer(liveConf, true); + + liveJMSService = new JMSServerManagerImpl(liveService); + + liveJMSService.setContext(ctx1); + + liveJMSService.start(); + + } + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java 2010-03-28 21:01:50 UTC (rev 8987) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java 2010-03-29 02:53:39 UTC (rev 8988) @@ -35,6 +35,7 @@ import org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory; import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; import org.hornetq.core.remoting.impl.invm.TransportConstants; +import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.HornetQServers; import org.hornetq.jms.client.HornetQConnectionFactory; @@ -599,6 +600,20 @@ { return false; //To change body of implemented methods use File | Settings | File Templates. } + + /* (non-Javadoc) + * @see org.hornetq.jms.persistence.JMSStorageManager#installReplication(org.hornetq.core.replication.ReplicationEndpoint) + */ + public void installReplication(ReplicationEndpoint replicationEndpoint) throws Exception + { + } + + /* (non-Javadoc) + * @see org.hornetq.jms.persistence.JMSStorageManager#load() + */ + public void load() throws Exception + { + } } } \ No newline at end of file