Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-3119 - proxy connector and failover 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1055873 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2011-01-06 13:44:31 +00:00
parent 6c721fcff5
commit 86565eb0e1
3 changed files with 245 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
activemq-core/src/test/java/org/apache/activemq/proxy/ProxyFailoverTest.java Normal file
View File

@ -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();
}
}

69
activemq-core/src/test/java/org/apache/activemq/util/ConsumerThread.java Normal file
View File

@ -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;
}
}

73
activemq-core/src/test/java/org/apache/activemq/util/ProducerThread.java Normal file
View File

@ -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;
}
}