activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dej...@apache.org
Subject svn commit: r1055873 - in /activemq/trunk/activemq-core/src/test/java/org/apache/activemq: proxy/ProxyFailoverTest.java util/ConsumerThread.java util/ProducerThread.java
Date Thu, 06 Jan 2011 13:44:32 GMT
Author: dejanb
Date: Thu Jan  6 13:44:31 2011
New Revision: 1055873

URL: http://svn.apache.org/viewvc?rev=1055873&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-3119 - proxy connector and failover

Added:
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/proxy/ProxyFailoverTest.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ConsumerThread.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ProducerThread.java

Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/proxy/ProxyFailoverTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/proxy/ProxyFailoverTest.java?rev=1055873&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/proxy/ProxyFailoverTest.java
(added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/proxy/ProxyFailoverTest.java
Thu Jan  6 13:44:31 2011
@@ -0,0 +1,103 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activemq.proxy;
+
+import junit.framework.TestCase;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.util.ConsumerThread;
+import org.apache.activemq.util.ProducerThread;
+
+import javax.jms.Connection;
+import javax.jms.Session;
+import java.net.URI;
+
+public class ProxyFailoverTest extends TestCase {
+
+    BrokerService proxyBroker;
+    BrokerService remoteBroker;
+
+    @Override
+    protected void setUp() throws Exception {
+
+        startRemoteBroker(true);
+
+        proxyBroker = new BrokerService();
+        ProxyConnector connector = new ProxyConnector();
+        connector.setBind(new URI("tcp://localhost:51618"));
+        connector.setProxyToLocalBroker(false);
+        connector.setRemote(new URI("failover:(tcp://localhost:61616)"));
+        proxyBroker.addProxyConnector(connector);
+        proxyBroker.setPersistent(false);
+        proxyBroker.setUseJmx(false);
+        proxyBroker.start();
+        proxyBroker.waitUntilStarted();
+
+    }
+
+    @Override
+    protected void tearDown() throws Exception {
+        proxyBroker.stop();
+        proxyBroker.waitUntilStopped();
+
+        remoteBroker.stop();
+        remoteBroker.waitUntilStopped();
+    }
+
+    public void testFailover() throws Exception {
+        ActiveMQConnectionFactory producerFactory = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616,tcp://localhost:61626)?randomize=false");
+        Connection producerConnection = producerFactory.createConnection();
+        producerConnection.start();
+        Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        ProducerThread producer = new ProducerThread(producerSession, producerSession.createQueue("ProxyTest"));
+        producer.setSleep(10);
+        producer.start();
+
+
+        ActiveMQConnectionFactory consumerFactory = new ActiveMQConnectionFactory("tcp://localhost:51618?wireFormat.cacheEnabled=false");
+        // if used with cached enabled it will fail
+        //ActiveMQConnectionFactory consumerFactory = new ActiveMQConnectionFactory("tcp://localhost:51618");
+        Connection consumerConnection = consumerFactory.createConnection();
+        consumerConnection.start();
+        Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        ConsumerThread consumer = new ConsumerThread(consumerSession, consumerSession.createQueue("ProxyTest"));
+        consumer.start();
+
+        Thread.sleep(5000);
+
+        remoteBroker.stop();
+        remoteBroker.waitUntilStopped();
+        startRemoteBroker(false);
+
+        producer.join();
+        consumer.join();
+
+        assertEquals(1000, consumer.getReceived());
+
+    }
+
+    protected void startRemoteBroker(boolean delete) throws Exception {
+        remoteBroker = new BrokerService();
+        remoteBroker.addConnector("tcp://localhost:61616");
+        if (delete) {
+            remoteBroker.deleteAllMessages();
+        }
+        remoteBroker.setUseJmx(false);
+        remoteBroker.start();
+        remoteBroker.waitUntilStarted();
+    }
+}

Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ConsumerThread.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ConsumerThread.java?rev=1055873&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ConsumerThread.java
(added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ConsumerThread.java
Thu Jan  6 13:44:31 2011
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activemq.util;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.jms.*;
+
+public class ConsumerThread extends Thread {
+
+    private static final Log LOG = LogFactory.getLog(ConsumerThread.class);
+
+    int messageCount = 1000;
+    int received = 0;
+    Destination dest;
+    Session sess;
+
+    public ConsumerThread(Session sess, Destination dest) {
+        this.dest = dest;
+        this.sess = sess;
+    }
+
+    @Override
+    public void run() {
+      MessageConsumer consumer = null;
+
+        try {
+            consumer = sess.createConsumer(dest);
+            while (received < messageCount) {
+                Message msg = consumer.receive(3000);
+                if (msg != null) {
+                    LOG.info("Received " + ((TextMessage)msg).getText());
+                    received++;
+                } else {
+                    break;
+                }
+            }
+        } catch (JMSException e) {
+            e.printStackTrace();
+        } finally {
+            if (consumer != null) {
+                try {
+                    consumer.close();
+                } catch (JMSException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+    public int getReceived() {
+        return received;
+    }
+}

Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ProducerThread.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ProducerThread.java?rev=1055873&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ProducerThread.java
(added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/util/ProducerThread.java
Thu Jan  6 13:44:31 2011
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activemq.util;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+public class ProducerThread extends Thread {
+
+    private static final Log LOG = LogFactory.getLog(ProducerThread.class);
+
+    int messageCount = 1000;
+    Destination dest;
+    Session sess;
+    int sleep = 0;
+
+    public ProducerThread(Session sess, Destination dest) {
+        this.dest = dest;
+        this.sess = sess;
+    }
+
+    public void run() {
+        MessageProducer producer = null;
+        try {
+            producer = sess.createProducer(dest);
+            for (int i = 0; i < messageCount; i++) {
+                producer.send(sess.createTextMessage("test message: " + i));
+                LOG.info("Sent 'test message: " + i + "'");
+                if (sleep > 0) {
+                    Thread.sleep(sleep);
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            if (producer != null) {
+                try {
+                    producer.close();
+                } catch (JMSException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+
+    public void setMessageCount(int messageCount) {
+        this.messageCount = messageCount;
+    }
+
+    public void setSleep(int sleep) {
+        this.sleep = sleep;
+    }
+}



Mime
View raw message