[hornetq-commits] JBoss hornetq SVN: r8922 - in trunk: src/main/org/hornetq/jms/server/impl and 5 other directories.

Author: clebert.suconic(a)jboss.com Date: 2010-03-10 16:50:10 -0500 (Wed, 10 Mar 2010) New Revision: 8922 Added: trunk/tests/src/org/hornetq/tests/integration/jms/client/NoLocalSubscriberTest.java trunk/tests/src/org/hornetq/tests/integration/jms/cluster/TopicClusterTest.java trunk/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java Removed: trunk/tests/jms-tests/src/org/hornetq/jms/tests/NoLocalSubscriberTest.java Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java trunk/tests/src/org/hornetq/tests/unit/core/asyncio/AsynchronousFileTest.java Log: Fixing delete Topic & fixing the JMS Testsuite Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-03-10 21:09:19 UTC (rev 8921) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-03-10 21:50:10 UTC (rev 8922) @@ -691,7 +691,7 @@ if (queue.getConsumerCount() != 0) { - throw new HornetQException(HornetQException.ILLEGAL_STATE, "Cannot delete queue - it has consumers"); + throw new HornetQException(HornetQException.ILLEGAL_STATE, "Cannot delete queue " + queue.getName() + " on binding " + queueName + " - it has consumers = " + binding.getClass().getName()); } if (session != null) Modified: trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-03-10 21:09:19 UTC (rev 8921) +++ trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-03-10 21:50:10 UTC (rev 8922) @@ -27,6 +27,7 @@ import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.Pair; +import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.management.AddressControl; import org.hornetq.api.core.management.ResourceNames; @@ -37,6 +38,8 @@ import org.hornetq.core.deployers.impl.FileDeploymentManager; import org.hornetq.core.deployers.impl.XmlDeployer; import org.hornetq.core.logging.Logger; +import org.hornetq.core.postoffice.Binding; +import org.hornetq.core.postoffice.BindingType; import org.hornetq.core.server.ActivateCallback; import org.hornetq.core.server.HornetQServer; import org.hornetq.jms.client.HornetQConnectionFactory; @@ -353,7 +356,18 @@ { for (String queueName : addressControl.getQueueNames()) { - server.getHornetQServerControl().destroyQueue(queueName); + Binding binding = server.getPostOffice().getBinding(new SimpleString(queueName)); + if (binding == null) + { + log.warn("Queue " + queueName + " doesn't exist on the topic " + name + ". It was deleted manually probably."); + continue; + } + + // We can't remove the remote binding. As this would be the bridge associated with the topic on this case + if (binding.getType() != BindingType.REMOTE_QUEUE) + { + server.getHornetQServerControl().destroyQueue(queueName); + } } } return true; Deleted: trunk/tests/jms-tests/src/org/hornetq/jms/tests/NoLocalSubscriberTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/NoLocalSubscriberTest.java 2010-03-10 21:09:19 UTC (rev 8921) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/NoLocalSubscriberTest.java 2010-03-10 21:50:10 UTC (rev 8922) @@ -1,118 +0,0 @@ -/* - * 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.jms.tests; - -import javax.jms.Connection; -import javax.jms.Message; -import javax.jms.MessageConsumer; -import javax.jms.MessageProducer; -import javax.jms.Session; -import javax.jms.TextMessage; - -import org.hornetq.tests.util.RandomUtil; - -/** - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - * - */ -public class NoLocalSubscriberTest extends JMSTestCase -{ - // Constants ----------------------------------------------------- - - // Static -------------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - /** - * Test that a message created from the same connection than a nolocal consumer - * can be sent by *another* connection and will be received by the nolocal consumer - */ - public void testNoLocal() throws Exception - { - if (log.isTraceEnabled()) - { - log.trace("testNoLocal"); - } - - Connection defaultConn = null; - Connection newConn = null; - - try - { - defaultConn = JMSTestCase.cf.createConnection(); - Session defaultSess = defaultConn.createSession(false, Session.AUTO_ACKNOWLEDGE); - MessageConsumer defaultConsumer = defaultSess.createConsumer(HornetQServerTestCase.topic1); - MessageConsumer noLocalConsumer = defaultSess.createConsumer(HornetQServerTestCase.topic1, null, true); - MessageProducer defaultProd = defaultSess.createProducer(HornetQServerTestCase.topic1); - - defaultConn.start(); - - String text = RandomUtil.randomString(); - // message is created only once from the same connection than the noLocalConsumer - TextMessage messageSent = defaultSess.createTextMessage(text); - for (int i = 0; i < 10; i++) - { - defaultProd.send(messageSent); - } - - Message received = null; - for (int i = 0; i < 10; i++) - { - received = defaultConsumer.receive(5000); - assertNotNull(received); - assertEquals(text, ((TextMessage)received).getText()); - } - - newConn = JMSTestCase.cf.createConnection(); - Session newSession = newConn.createSession(false, Session.AUTO_ACKNOWLEDGE); - MessageProducer newProd = newSession.createProducer(HornetQServerTestCase.topic1); - MessageConsumer newConsumer = newSession.createConsumer(HornetQServerTestCase.topic1); - - newConn.start(); - - text = RandomUtil.randomString(); - messageSent.setText(text); - defaultProd.send(messageSent); - - received = newConsumer.receive(5000); - assertNotNull(received); - assertEquals(text, ((TextMessage)received).getText()); - - text = RandomUtil.randomString(); - messageSent.setText(text); - // we send the message created at the start of the test but on the *newConn* this time - newProd.send(messageSent); - newConn.close(); - - received = noLocalConsumer.receive(5000); - assertNotNull("nolocal consumer did not get message", received); - assertEquals(text, ((TextMessage)received).getText()); - } - finally - { - if (defaultConn != null) - { - defaultConn.close(); - } - if (newConn != null) - { - newConn.close(); - } - } - } -} Copied: trunk/tests/src/org/hornetq/tests/integration/jms/client/NoLocalSubscriberTest.java (from rev 8920, trunk/tests/jms-tests/src/org/hornetq/jms/tests/NoLocalSubscriberTest.java) =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/client/NoLocalSubscriberTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/jms/client/NoLocalSubscriberTest.java 2010-03-10 21:50:10 UTC (rev 8922) @@ -0,0 +1,125 @@ +/* + * 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.client; + +import javax.jms.Connection; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; +import javax.jms.Topic; + +import org.hornetq.core.logging.Logger; +import org.hornetq.tests.util.JMSTestBase; +import org.hornetq.tests.util.RandomUtil; + +/** + * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> + * + */ +public class NoLocalSubscriberTest extends JMSTestBase +{ + // Constants ----------------------------------------------------- + + private static final Logger log = Logger.getLogger(NoLocalSubscriberTest.class); + + + // Static -------------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + /** + * Test that a message created from the same connection than a nolocal consumer + * can be sent by *another* connection and will be received by the nolocal consumer + */ + public void testNoLocal() throws Exception + { + if (log.isTraceEnabled()) + { + log.trace("testNoLocal"); + } + + Connection defaultConn = null; + Connection newConn = null; + + try + { + Topic topic1 = createTopic("topic1"); + defaultConn = cf.createConnection(); + Session defaultSess = defaultConn.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageConsumer defaultConsumer = defaultSess.createConsumer(topic1); + MessageConsumer noLocalConsumer = defaultSess.createConsumer(topic1, null, true); + MessageProducer defaultProd = defaultSess.createProducer(topic1); + + defaultConn.start(); + + String text = RandomUtil.randomString(); + // message is created only once from the same connection than the noLocalConsumer + TextMessage messageSent = defaultSess.createTextMessage(text); + for (int i = 0; i < 10; i++) + { + defaultProd.send(messageSent); + } + + Message received = null; + for (int i = 0; i < 10; i++) + { + received = defaultConsumer.receive(5000); + assertNotNull(received); + assertEquals(text, ((TextMessage)received).getText()); + } + + newConn = cf.createConnection(); + Session newSession = newConn.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageProducer newProd = newSession.createProducer(topic1); + MessageConsumer newConsumer = newSession.createConsumer(topic1); + + newConn.start(); + + text = RandomUtil.randomString(); + messageSent.setText(text); + defaultProd.send(messageSent); + + received = newConsumer.receive(5000); + assertNotNull(received); + assertEquals(text, ((TextMessage)received).getText()); + + text = RandomUtil.randomString(); + messageSent.setText(text); + // we send the message created at the start of the test but on the *newConn* this time + newProd.send(messageSent); + newConn.close(); + + received = noLocalConsumer.receive(5000); + assertNotNull("nolocal consumer did not get message", received); + assertEquals(text, ((TextMessage)received).getText()); + } + finally + { + if (defaultConn != null) + { + defaultConn.close(); + } + if (newConn != null) + { + newConn.close(); + } + } + } +} Added: trunk/tests/src/org/hornetq/tests/integration/jms/cluster/TopicClusterTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/cluster/TopicClusterTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/jms/cluster/TopicClusterTest.java 2010-03-10 21:50:10 UTC (rev 8922) @@ -0,0 +1,125 @@ +/* + * 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 javax.jms.Connection; +import javax.jms.DeliveryMode; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; +import javax.jms.Topic; + +import org.hornetq.tests.util.JMSClusteredTestBase; + +/** + * A TopicClusterTest + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class TopicClusterTest extends JMSClusteredTestBase +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + + protected void tearDown() throws Exception + { + super.tearDown(); + } + + + protected void setUp() throws Exception + { + super.setUp(); + } + + public void testDeleteTopicAfterClusteredSend() throws Exception + { + Connection conn1 = cf1.createConnection(); + + conn1.setClientID("someClient1"); + + Connection conn2 = cf2.createConnection(); + + conn2.setClientID("someClient2"); + + conn1.start(); + + conn2.start(); + + try + { + + Topic topic1 = createTopic("t1"); + + Topic topic2 = (Topic)context1.lookup("topic/t1"); + + Session session1 = conn1.createSession(false, Session.AUTO_ACKNOWLEDGE); + + Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE); + + // topic1 and 2 should be the same. + // Using a different instance here just to make sure it is implemented correctly + MessageConsumer cons2 = session2.createDurableSubscriber(topic2, "sub2"); + + MessageProducer prod1 = session1.createProducer(topic1); + + prod1.setDeliveryMode(DeliveryMode.PERSISTENT); + + + for (int i = 0 ; i < 1000; i++) + { + prod1.send(session1.createTextMessage("someMessage")); + } + + TextMessage received = (TextMessage)cons2.receive(5000); + + assertNotNull(received); + + assertEquals("someMessage", received.getText()); + + cons2.close(); + } + finally + { + conn1.close(); + conn2.close(); + } + + jmsServer1.destroyTopic("t1"); + jmsServer2.destroyTopic("t1"); + + + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/tests/src/org/hornetq/tests/unit/core/asyncio/AsynchronousFileTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/asyncio/AsynchronousFileTest.java 2010-03-10 21:09:19 UTC (rev 8921) +++ trunk/tests/src/org/hornetq/tests/unit/core/asyncio/AsynchronousFileTest.java 2010-03-10 21:50:10 UTC (rev 8922) @@ -290,7 +290,7 @@ callbackLocal.latch.await(); - // assertTrue(callbackLocal.error); + assertTrue(callbackLocal.error); callbackLocal = new LocalCallback(); Added: trunk/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/util/JMSClusteredTestBase.java 2010-03-10 21:50:10 UTC (rev 8922) @@ -0,0 +1,270 @@ +/* + * 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.util; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import javax.jms.ConnectionFactory; +import javax.jms.Queue; +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; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory; +import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.HornetQServers; +import org.hornetq.jms.server.config.impl.JMSConfigurationImpl; +import org.hornetq.jms.server.config.impl.TopicConfigurationImpl; +import org.hornetq.jms.server.impl.JMSServerManagerImpl; +import org.hornetq.tests.integration.cluster.distribution.ClusterTestBase; +import org.hornetq.tests.unit.util.InVMContext; + +/** + * A JMSBaseTest + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class JMSClusteredTestBase extends ServiceTestBase +{ + + private static final Logger log = Logger.getLogger(ClusterTestBase.class); + + protected HornetQServer server1; + + protected JMSServerManagerImpl jmsServer1; + + protected HornetQServer server2; + + protected JMSServerManagerImpl jmsServer2; + + protected ConnectionFactory cf1; + + protected ConnectionFactory cf2; + + protected InVMContext context1; + + protected InVMContext context2; + + private static final int MAX_HOPS = 1; + + // Static -------------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Constructors -------------------------------------------------- + + // TestCase overrides ------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + /** + * @throws Exception + * @throws NamingException + */ + protected Queue createQueue(final String name) throws Exception, NamingException + { + jmsServer2.createQueue(name, "/queue/" + name, null, true); + jmsServer1.createQueue(name, "/queue/" + name, null, true); + + return (Queue)context1.lookup("/queue/" + name); + } + + protected Topic createTopic(final String name) throws Exception, NamingException + { + jmsServer2.createTopic(name, "/topic/" + name); + jmsServer1.createTopic(name, "/topic/" + name); + + return (Topic)context1.lookup("/topic/" + name); + } + + @Override + protected void setUp() throws Exception + { + super.setUp(); + + setupServer2(); + setupServer1(); + + jmsServer2.start(); + jmsServer2.activated(); + + jmsServer1.start(); + jmsServer1.activated(); + + cf1 = HornetQJMSClient.createConnectionFactory(new TransportConfiguration(InVMConnectorFactory.class.getName(), + generateInVMParams(1))); + cf2 = HornetQJMSClient.createConnectionFactory(new TransportConfiguration(InVMConnectorFactory.class.getName(), + generateInVMParams(2))); + } + + /** + * @param toOtherServerPair + * @throws Exception + */ + private void setupServer2() throws Exception + { + List<Pair<String, String>> toOtherServerPair = new ArrayList<Pair<String, String>>(); + toOtherServerPair.add(new Pair<String, String>("toServer1", null)); + + Configuration conf2 = createDefaultConfig(1, generateInVMParams(2), InVMAcceptorFactory.class.getCanonicalName()); + conf2.setSecurityEnabled(false); + conf2.setJMXManagementEnabled(true); + conf2.setPersistenceEnabled(false); + + conf2.getConnectorConfigurations().put("toServer1", + new TransportConfiguration(InVMConnectorFactory.class.getName(), + generateInVMParams(1))); + + conf2.setClustered(true); + + conf2.getClusterConfigurations().add(new ClusterConnectionConfiguration("to-server1", + "jms", + 1000, + true, + true, + MAX_HOPS, + 1024, + toOtherServerPair)); + + + JMSConfigurationImpl jmsconfig = new JMSConfigurationImpl(); + //jmsconfig.getTopicConfigurations().add(new TopicConfigurationImpl("t1", "topic/t1")); + + + server2 = HornetQServers.newHornetQServer(conf2, false); + jmsServer2 = new JMSServerManagerImpl(server2, jmsconfig); + context2 = new InVMContext(); + jmsServer2.setContext(context2); + } + + /** + * @param toOtherServerPair + * @throws Exception + */ + private void setupServer1() throws Exception + { + List<Pair<String, String>> toOtherServerPair = new ArrayList<Pair<String, String>>(); + toOtherServerPair.add(new Pair<String, String>("toServer2", null)); + + Configuration conf1 = createDefaultConfig(1, generateInVMParams(1), InVMAcceptorFactory.class.getCanonicalName()); + + conf1.setSecurityEnabled(false); + conf1.setJMXManagementEnabled(true); + conf1.setPersistenceEnabled(false); + + conf1.getConnectorConfigurations().put("toServer2", + new TransportConfiguration(InVMConnectorFactory.class.getName(), + generateInVMParams(2))); + + // TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY, params); + + conf1.setClustered(true); + + conf1.getClusterConfigurations().add(new ClusterConnectionConfiguration("to-server2", + "jms", + 1000, + true, + true, + MAX_HOPS, + 1024, + toOtherServerPair)); + + + JMSConfigurationImpl jmsconfig = new JMSConfigurationImpl(); + //jmsconfig.getTopicConfigurations().add(new TopicConfigurationImpl("t1", "topic/t1")); + + server1 = HornetQServers.newHornetQServer(conf1, false); + jmsServer1 = new JMSServerManagerImpl(server1, jmsconfig); + context1 = new InVMContext(); + jmsServer1.setContext(context1); + } + + @Override + protected void tearDown() throws Exception + { + + try + { + jmsServer2.stop(); + + server2.stop(); + + context2.close(); + } + catch (Throwable e) + { + log.warn("Can't stop server2", e); + } + + server2 = null; + + jmsServer2 = null; + + context2 = null; + + cf1 = null; + + try + { + jmsServer1.stop(); + + server1.stop(); + + context1.close(); + } + catch (Throwable e) + { + log.warn("Can't stop server2", e); + } + + server1 = null; + + jmsServer1 = null; + + context1 = null; + + super.tearDown(); + } + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + + protected Map<String, Object> generateInVMParams(final int node) + { + Map<String, Object> params = new HashMap<String, Object>(); + + params.put(org.hornetq.core.remoting.impl.invm.TransportConstants.SERVER_ID_PROP_NAME, node); + + return params; + } + + +}