[no jira] static network wildcard drain test

This commit is contained in:
gtully 2016-11-11 11:49:56 +00:00
parent 097c0e7eae
commit 7d0df2d01e
1 changed files with 92 additions and 0 deletions

View File

@ -0,0 +1,92 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.network;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import javax.jms.Connection;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
public class DrainBridgeTest {
@org.junit.Test
public void testDrain() throws Exception {
prepareBrokerWithMessages();
BrokerService target = prepareDrainTarget();
BrokerService drainingBroker = new BrokerService();
drainingBroker.setBrokerName("HOST");
// add the draining bridge that subscribes to all queues and forwards on start - irrespective of demand
NetworkConnector drainingNetworkConnector = drainingBroker.addNetworkConnector("static:(" + target.getTransportConnectorByScheme("tcp").getPublishableConnectString() + ")");
drainingNetworkConnector.setStaticBridge(true);
drainingNetworkConnector.setStaticallyIncludedDestinations(Arrays.asList(new ActiveMQDestination[]{new ActiveMQQueue("*")}));
drainingBroker.start();
System.out.println("Local count: " + drainingBroker.getAdminView().getTotalMessageCount() + ", target count:" + target.getAdminView().getTotalMessageCount());
assertEquals("local messages", 20, drainingBroker.getAdminView().getTotalMessageCount());
assertEquals("no remote messages", 0, target.getAdminView().getTotalMessageCount());
while (drainingBroker.getAdminView().getTotalMessageCount() > 0) {
TimeUnit.SECONDS.sleep(5);
}
assertEquals("no local messages", 0, drainingBroker.getAdminView().getTotalMessageCount());
assertEquals("remote messages", 20, target.getAdminView().getTotalMessageCount());
drainingBroker.stop();
target.stop();
}
private BrokerService prepareDrainTarget() throws Exception {
BrokerService broker = new BrokerService();
broker.setDeleteAllMessagesOnStartup(true);
broker.setBrokerName("TARGET");
broker.addConnector("tcp://localhost:0");
broker.start();
return broker;
}
private void prepareBrokerWithMessages() throws Exception {
BrokerService broker = new BrokerService();
broker.setDeleteAllMessagesOnStartup(true);
broker.setBrokerName("HOST");
broker.start();
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getVmConnectorURI());
Connection conn = connectionFactory.createConnection();
conn.start();
Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
TextMessage msg = session.createTextMessage("This is a message.");
MessageProducer producer = session.createProducer(new ActiveMQQueue("Foo,Bar"));
for (int i = 0; i < 10; i++) {
producer.send(msg);
}
conn.close();
broker.stop();
}
}