Added additional test case for multicasted broker.

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@374637 13f79535-47bb-0310-9956-ffa450edef68

assembly/src/test/java/org/apache/activemq/usecases/TwoBrokerMulticastQueueTest.java

@@ -0,0 +1,269 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.apache.activemq.usecases;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.CombinationTestSupport;
+import org.apache.activemq.util.MessageIdList;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.xbean.XBeanBrokerFactory;
+
+import java.net.URI;
+import java.util.Arrays;
+
+import junit.framework.Test;
+
+import javax.jms.Destination;
+import javax.jms.ConnectionFactory;
+import javax.jms.Connection;
+import javax.jms.Session;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.TextMessage;
+
+public class TwoBrokerMulticastQueueTest extends CombinationTestSupport {
+
+    public static Test suite() {
+        return suite(TwoBrokerMulticastQueueTest.class);
+    }
+
+    public static void main(String[] args) {
+        junit.textui.TestRunner.run(suite());
+    }
+
+    public static final int MESSAGE_COUNT  = 100;
+    public static final int BROKER_COUNT   = 2;
+    public static final int CONSUMER_COUNT = 20;
+
+    private BrokerService[] brokers;
+    public String sendUri, recvUri;
+
+    public void setUp() throws Exception {
+        super.setAutoFail(true);
+        super.setUp();
+    }
+
+    public void tearDown() throws Exception {
+        for (int i=0; i<BROKER_COUNT; i++) {
+            brokers[i].stop();
+        }
+        super.tearDown();
+    }
+
+    private void doSendReceiveTest() throws Exception {
+        Destination dest = new ActiveMQQueue("TEST.FOO");
+
+        ConnectionFactory sendFactory = createConnectionFactory(sendUri);
+
+        Connection conn = createConnection(sendFactory);
+        sendMessages(conn, dest, MESSAGE_COUNT);
+
+        Thread.sleep(500);
+
+        ConnectionFactory recvFactory = createConnectionFactory(recvUri);
+        assertEquals(MESSAGE_COUNT, receiveMessages(createConnection(recvFactory), dest, 0));
+    }
+
+    private void doMultipleConsumersConnectTest() throws Exception {
+        Destination dest = new ActiveMQQueue("TEST.FOO");
+
+        ConnectionFactory sendFactory = createConnectionFactory(sendUri);
+
+        Connection conn = createConnection(sendFactory);
+        sendMessages(conn, dest, MESSAGE_COUNT);
+
+        Thread.sleep(500);
+
+        ConnectionFactory recvFactory = createConnectionFactory(recvUri);
+        assertEquals(MESSAGE_COUNT, receiveMessages(createConnection(recvFactory), dest, 0));
+
+        for (int i=0; i<(CONSUMER_COUNT-1); i++) {
+            assertEquals(0, receiveMessages(createConnection(recvFactory), dest, 200));
+        }
+    }
+
+    public void initCombosForTestSendReceive() {
+        addCombinationValues("sendUri", new Object[] {
+            "tcp://localhost:61616", "tcp://localhost:61617"
+        });
+        addCombinationValues("recvUri", new Object[] {
+            "tcp://localhost:61616", "tcp://localhost:61617"
+        });
+    }
+
+    public void testSendReceive() throws Throwable {
+        createMulticastBrokerNetwork();
+        doSendReceiveTest();
+    }
+
+    public void initCombosForTestMultipleConsumersConnect() {
+        addCombinationValues("sendUri", new Object[] {
+            "tcp://localhost:61616", "tcp://localhost:61617",
+        });
+        addCombinationValues("recvUri", new Object[] {
+            "tcp://localhost:61616", "tcp://localhost:61617"
+        });
+    }
+
+    public void testMultipleConsumersConnect() throws Throwable {
+        createMulticastBrokerNetwork();
+        doMultipleConsumersConnectTest();
+    }
+
+    public void testSendReceiveUsingFailover() throws Throwable {
+        sendUri = "failover:tcp://localhost:61616,tcp://localhost:61617";
+        recvUri = "failover:tcp://localhost:61616,tcp://localhost:61617";
+        createMulticastBrokerNetwork();
+        doSendReceiveTest();
+    }
+
+    public void testMultipleConsumersConnectUsingFailover() throws Throwable {
+        sendUri = "failover:tcp://localhost:61616,tcp://localhost:61617";
+        recvUri = "failover:tcp://localhost:61616,tcp://localhost:61617";
+        createMulticastBrokerNetwork();
+        doMultipleConsumersConnectTest();
+    }
+
+    public void testSendReceiveUsingDiscovery() throws Throwable {
+        sendUri = "discovery:multicast://default";
+        recvUri = "discovery:multicast://default";
+        createMulticastBrokerNetwork();
+        doSendReceiveTest();
+    }
+
+    public void testMultipleConsumersConnectUsingDiscovery() throws Throwable {
+        sendUri = "discovery:multicast://default";
+        recvUri = "discovery:multicast://default";
+        createMulticastBrokerNetwork();
+        doMultipleConsumersConnectTest();
+    }
+
+    public void testSendReceiveUsingAutoAssignFailover() throws Throwable {
+        sendUri = "failover:multicast://default";
+        recvUri = "failover:multicast://default";
+        createAutoAssignMulticastBrokerNetwork();
+        doSendReceiveTest();
+    }
+
+    public void testMultipleConsumersConnectUsingAutoAssignFailover() throws Throwable {
+        sendUri = "failover:multicast://default";
+        recvUri = "failover:multicast://default";
+        createAutoAssignMulticastBrokerNetwork();
+        doMultipleConsumersConnectTest();
+    }
+
+    public void testSendReceiveUsingAutoAssignDiscovery() throws Throwable {
+        sendUri = "discovery:multicast://default";
+        recvUri = "discovery:multicast://default";
+        createAutoAssignMulticastBrokerNetwork();
+        doSendReceiveTest();
+    }
+
+    public void testMultipleConsumersConnectUsingAutoAssignDiscovery() throws Throwable {
+        sendUri = "discovery:multicast://default";
+        recvUri = "discovery:multicast://default";
+        createAutoAssignMulticastBrokerNetwork();
+        doMultipleConsumersConnectTest();
+    }
+
+    protected void createMulticastBrokerNetwork() throws Exception {
+
+        brokers = new BrokerService[BROKER_COUNT];
+        for (int i=0; i<BROKER_COUNT; i++) {
+            brokers[i] = createBroker("xbean:multicast-broker-" + (i+1)+ ".xml");
+            brokers[i].start();
+        }
+
+        // Let the brokers discover each other first
+        Thread.sleep(1000);
+    }
+
+    protected void createAutoAssignMulticastBrokerNetwork() throws Exception {
+        brokers = new BrokerService[BROKER_COUNT];
+        for (int i=0; i<BROKER_COUNT; i++) {
+            brokers[i] = createBroker("xbean:multicast-broker-auto.xml");
+            brokers[i].start();
+        }
+
+        // Let the brokers discover each other first
+        Thread.sleep(1000);
+    }
+
+    protected BrokerService createBroker(String uri) throws Exception {
+        return (new XBeanBrokerFactory()).createBroker(new URI(uri));
+    }
+
+    protected ConnectionFactory createConnectionFactory(String uri) {
+        return new ActiveMQConnectionFactory(uri);
+    }
+
+    protected Connection createConnection(ConnectionFactory factory) throws JMSException {
+        Connection conn = factory.createConnection();
+        return conn;
+    }
+
+    protected int receiveMessages(Connection conn, Destination dest, int waitTime) throws JMSException, InterruptedException {
+        conn.start();
+        MessageIdList list = new MessageIdList();
+        Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageConsumer consumer = sess.createConsumer(dest);
+        consumer.setMessageListener(list);
+
+        if (waitTime > 0) {
+            Thread.sleep(waitTime);
+        } else {
+            list.waitForMessagesToArrive(MESSAGE_COUNT);
+        }
+
+        conn.close();
+
+        return list.getMessageCount();
+    }
+
+    protected void sendMessages(Connection conn, Destination dest, int count) throws JMSException {
+        conn.start();
+        Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer prod = sess.createProducer(dest);
+
+        for (int i=0; i<count; i++) {
+            prod.send(createTextMessage(sess, "Message " + i, 1024));
+        }
+
+        conn.close();
+    }
+
+    protected TextMessage createTextMessage(Session session, String initText, int messageSize) throws JMSException {
+        TextMessage msg = session.createTextMessage();
+
+        // Pad message text
+        if (initText.length() < messageSize) {
+            char[] data = new char[messageSize - initText.length()];
+            Arrays.fill(data, '*');
+            String str = new String(data);
+            msg.setText(initText + str);
+
+        // Do not pad message text
+        } else {
+            msg.setText(initText);
+        }
+
+        return msg;
+    }
+
+}

assembly/src/test/resources/org/apache/activemq/usecases/multicast-broker-1.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Copyright 2005-2006 The Apache Software Foundation
+   
+    Licensed 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.
+-->
+<beans xmlns="http://activemq.org/config/1.0">
+
+  <broker brokerName="BrokerA" persistent="false" useJmx="false">
+    <transportConnectors>
+      <transportConnector uri="tcp://localhost:61616" discoveryUri="multicast://default"/>
+    </transportConnectors>
+
+    <networkConnectors>
+      <networkConnector uri="multicast://default"/>
+    </networkConnectors>
+
+    <persistenceAdapter>
+      <memoryPersistenceAdapter/>
+    </persistenceAdapter>
+  </broker>
+
+</beans>

assembly/src/test/resources/org/apache/activemq/usecases/multicast-broker-2.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Copyright 2005-2006 The Apache Software Foundation
+   
+    Licensed 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.
+-->
+<beans xmlns="http://activemq.org/config/1.0">
+
+  <broker brokerName="BrokerB" persistent="false" useJmx="false">
+    <transportConnectors>
+      <transportConnector uri="tcp://localhost:61617" discoveryUri="multicast://default"/>
+    </transportConnectors>
+
+    <networkConnectors>
+      <networkConnector uri="multicast://default"/>
+    </networkConnectors>
+
+    <persistenceAdapter>
+      <memoryPersistenceAdapter/>
+    </persistenceAdapter>
+  </broker>
+
+</beans>

assembly/src/test/resources/org/apache/activemq/usecases/multicast-broker-auto.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Copyright 2005-2006 The Apache Software Foundation
+   
+    Licensed 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.
+-->
+<beans xmlns="http://activemq.org/config/1.0">
+
+  <broker brokerName="BrokerB" persistent="false" useJmx="false">
+    <transportConnectors>
+      <transportConnector uri="tcp://localhost:0" discoveryUri="multicast://default"/>
+    </transportConnectors>
+
+    <networkConnectors>
+      <networkConnector uri="multicast://default"/>
+    </networkConnectors>
+
+    <persistenceAdapter>
+      <memoryPersistenceAdapter/>
+    </persistenceAdapter>
+  </broker>
+
+</beans>