Author: clebert.suconic(a)jboss.com Date: 2010-11-25 00:37:49 -0500 (Thu, 25 Nov 2010) New Revision: 9926 Modified: branches/2_2_0_HA_Improvements_preMerge/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/jms/cluster/TemporaryQueueClusterTest.java branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java Log: Fixing a test Modified: branches/2_2_0_HA_Improvements_preMerge/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java =================================================================== --- branches/2_2_0_HA_Improvements_preMerge/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2010-11-24 04:25:38 UTC (rev 9925) +++ branches/2_2_0_HA_Improvements_preMerge/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2010-11-25 05:37:49 UTC (rev 9926) @@ -442,7 +442,7 @@ managementService.getManagementAddress(), managementService.getManagementNotificationAddress(), record, - connector); + this.connector); record.setBridge(bridge); Modified: branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java =================================================================== --- branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2010-11-24 04:25:38 UTC (rev 9925) +++ branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2010-11-25 05:37:49 UTC (rev 9926) @@ -27,10 +27,17 @@ import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.*; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ServerLocatorImpl; -import org.hornetq.core.config.*; +import org.hornetq.api.core.client.ClientConsumer; +import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientProducer; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.api.core.client.ServerLocator; +import org.hornetq.core.config.BroadcastGroupConfiguration; +import org.hornetq.core.config.ClusterConnectionConfiguration; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.DiscoveryGroupConfiguration; import org.hornetq.core.config.impl.ConfigurationImpl; import org.hornetq.core.logging.Logger; import org.hornetq.core.postoffice.Binding; @@ -1262,80 +1269,9 @@ protected void setupServer(final int node, final boolean fileStorage, final boolean netty) { - setupLiveServer(node, fileStorage, true, netty); + setupLiveServer(node, fileStorage, false, netty); } - /*protected void setupServer(final int node, - final boolean fileStorage, - final boolean sharedStorage, - final boolean netty, - final boolean backup, - final int backupNode, - final boolean useFakeLock) - { - if (servers[node] != null) - { - throw new IllegalArgumentException("Already a server at node " + node); - } - - Configuration configuration = new ConfigurationImpl(); - - configuration.setSecurityEnabled(false); - configuration.setJournalMinFiles(2); - configuration.setJournalMaxIO_AIO(1000); - configuration.setJournalFileSize(100 * 1024); - configuration.setJournalType(getDefaultJournalType()); - configuration.setSharedStore(sharedStorage); - if (sharedStorage && backup) - { - // Shared storage will share the node between the backup and live node - int nodeDirectoryToUse = backupNode == -1 ? node : backupNode; - configuration.setBindingsDirectory(getBindingsDir(nodeDirectoryToUse, false)); - configuration.setJournalDirectory(getJournalDir(nodeDirectoryToUse, false)); - configuration.setPagingDirectory(getPageDir(nodeDirectoryToUse, false)); - configuration.setLargeMessagesDirectory(getLargeMessagesDir(nodeDirectoryToUse, false)); - } - else - { - configuration.setBindingsDirectory(getBindingsDir(node, backup)); - configuration.setJournalDirectory(getJournalDir(node, backup)); - configuration.setPagingDirectory(getPageDir(node, backup)); - configuration.setLargeMessagesDirectory(getLargeMessagesDir(node, backup)); - } - configuration.setClustered(true); - configuration.setJournalCompactMinFiles(0); - configuration.setBackup(backup); - - configuration.getAcceptorConfigurations().clear(); - configuration.getAcceptorConfigurations().add(createTransportConfiguration(netty, true, generateParams(node, netty))); - - HornetQServer server; - - if (fileStorage) - { - if (useFakeLock) - { - server = createFakeLockServer(true, configuration); - } - else - { - server = HornetQServers.newHornetQServer(configuration); - } - } - else - { - if (useFakeLock) - { - server = createFakeLockServer(false, configuration); - } - else - { - server = HornetQServers.newHornetQServer(configuration, false); - } - } - servers[node] = server; - }*/ - protected void setupLiveServer(final int node, final boolean fileStorage, final boolean sharedStorage, Modified: branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/jms/cluster/TemporaryQueueClusterTest.java =================================================================== --- branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/jms/cluster/TemporaryQueueClusterTest.java 2010-11-24 04:25:38 UTC (rev 9925) +++ branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/integration/jms/cluster/TemporaryQueueClusterTest.java 2010-11-25 05:37:49 UTC (rev 9926) @@ -14,14 +14,11 @@ package org.hornetq.tests.integration.jms.cluster; import javax.jms.Connection; -import javax.jms.DeliveryMode; import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; -import javax.jms.TemporaryQueue; import javax.jms.TextMessage; -import javax.jms.Topic; import org.hornetq.tests.util.JMSClusteredTestBase; @@ -54,7 +51,56 @@ { super.setUp(); } + + + public void testClusteredQueue() throws Exception + { + System.out.println("Server1 = " + server1.getNodeID()); + System.out.println("Server2 = " + server2.getNodeID()); + jmsServer1.createQueue(false, "target", null, true, "/queue/target"); + jmsServer2.createQueue(false, "target", null, true, "/queue/target"); + + Connection conn1 = cf1.createConnection(); + Connection conn2 = cf2.createConnection(); + + conn1.start(); + + conn2.start(); + + try + { + Session session1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue targetQueue1 = session1.createQueue("target"); + + Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue targetQueue2 = session2.createQueue("target"); + + // sleep a little bit to have the temp queue propagated to server #2 + Thread.sleep(3000); + MessageProducer prod1 = session1.createProducer(targetQueue1); + MessageConsumer cons2 = session2.createConsumer(targetQueue2); + + TextMessage msg = session1.createTextMessage("hello"); + + prod1.send(msg); + + prod1.send(msg); + + TextMessage msgReceived = (TextMessage)cons2.receive(5000); + + assertNotNull(msgReceived); + assertEquals(msgReceived.getText(), msg.getText()); + + } + finally + { + conn1.close(); + conn2.close(); + } + } + + public void testTemporaryQueue() throws Exception { jmsServer1.createQueue(false, "target", null, false, "/queue/target"); @@ -75,11 +121,11 @@ Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue targetQueue2 = session2.createQueue("target"); - // sleep a little bit to have the temp queue propagated to server #2 - Thread.sleep(3000); MessageProducer prod1 = session1.createProducer(targetQueue1); MessageConsumer cons2 = session2.createConsumer(targetQueue2); MessageConsumer tempCons1 = session1.createConsumer(tempQueue); + // sleep a little bit to have the temp queue propagated to server #2 + Thread.sleep(3000); for (int i = 0; i < 10; i++) { Modified: branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java =================================================================== --- branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java 2010-11-24 04:25:38 UTC (rev 9925) +++ branches/2_2_0_HA_Improvements_preMerge/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java 2010-11-25 05:37:49 UTC (rev 9926) @@ -23,9 +23,7 @@ import javax.jms.Topic; import javax.naming.NamingException; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.HornetQClient; import org.hornetq.api.jms.HornetQJMSClient; import org.hornetq.core.config.ClusterConnectionConfiguration; import org.hornetq.core.config.Configuration; @@ -34,9 +32,8 @@ import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.HornetQServers; -import org.hornetq.jms.client.HornetQJMSConnectionFactory; +import org.hornetq.jms.client.HornetQConnectionFactory; import org.hornetq.jms.server.config.impl.JMSConfigurationImpl; -import org.hornetq.jms.server.config.impl.TopicConfigurationImpl; import org.hornetq.jms.server.impl.JMSFactoryType; import org.hornetq.jms.server.impl.JMSServerManagerImpl; import org.hornetq.tests.integration.cluster.distribution.ClusterTestBase; @@ -105,7 +102,7 @@ return (Topic)context1.lookup("/topic/" + name); } - + @Override protected void setUp() throws Exception { @@ -114,16 +111,16 @@ setupServer2(); setupServer1(); - jmsServer2.start(); - jmsServer2.activated(); - jmsServer1.start(); jmsServer1.activated(); - cf1 = (ConnectionFactory) HornetQJMSClient.createConnectionFactoryWithHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName(), + jmsServer2.start(); + jmsServer2.activated(); + + cf1 = (ConnectionFactory) HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName(), + generateInVMParams(0))); + cf2 = (ConnectionFactory)HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName(), generateInVMParams(1))); - cf2 = (ConnectionFactory)HornetQJMSClient.createConnectionFactoryWithHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName(), - generateInVMParams(2))); } /** @@ -135,14 +132,14 @@ List<String> toOtherServerPair = new ArrayList<String>(); toOtherServerPair.add("toServer1"); - Configuration conf2 = createDefaultConfig(2, generateInVMParams(2), InVMAcceptorFactory.class.getCanonicalName()); + Configuration conf2 = createDefaultConfig(1, generateInVMParams(1), InVMAcceptorFactory.class.getCanonicalName()); conf2.setSecurityEnabled(false); conf2.setJMXManagementEnabled(true); conf2.setPersistenceEnabled(false); conf2.getConnectorConfigurations().put("toServer1", new TransportConfiguration(InVMConnectorFactory.class.getName(), - generateInVMParams(1))); + generateInVMParams(0))); conf2.setClustered(true); @@ -176,7 +173,7 @@ List<String> toOtherServerPair = new ArrayList<String>(); toOtherServerPair.add("toServer2"); - Configuration conf1 = createDefaultConfig(1, generateInVMParams(1), InVMAcceptorFactory.class.getCanonicalName()); + Configuration conf1 = createDefaultConfig(0, generateInVMParams(0), InVMAcceptorFactory.class.getCanonicalName()); conf1.setSecurityEnabled(false); conf1.setJMXManagementEnabled(true); @@ -184,10 +181,8 @@ conf1.getConnectorConfigurations().put("toServer2", new TransportConfiguration(InVMConnectorFactory.class.getName(), - generateInVMParams(2))); + generateInVMParams(1))); - // TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY, params); - conf1.setClustered(true); conf1.getClusterConfigurations().add(new ClusterConnectionConfiguration("to-server2", @@ -226,6 +221,10 @@ { log.warn("Can't stop server2", e); } + + ((HornetQConnectionFactory)cf1).close(); + + ((HornetQConnectionFactory)cf2).close(); server2 = null;