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

attempt to fix test case after changes in https://issues.apache.org/jira/browse/AMQ-4237 broke the test's MBean lookup, not quite there but close. 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1429149 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish 2013-01-04 21:58:25 +00:00
parent b5625e31ee
commit 5a8ca0b8a8
1 changed files with 72 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

132
activemq-core/src/test/java/org/apache/activemq/transport/discovery/DiscoveryNetworkReconnectTest.java
View File

@ -30,8 +30,6 @@ import org.apache.activemq.broker.jmx.ManagementContext;
import org.apache.activemq.transport.discovery.multicast.MulticastDiscoveryAgentFactory; import org.apache.activemq.transport.discovery.multicast.MulticastDiscoveryAgentFactory;
import org.apache.activemq.util.SocketProxy; import org.apache.activemq.util.SocketProxy;
import org.apache.activemq.util.Wait; import org.apache.activemq.util.Wait;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.hamcrest.BaseMatcher; import org.hamcrest.BaseMatcher;
import org.hamcrest.Description; import org.hamcrest.Description;
import org.jmock.Expectations; import org.jmock.Expectations;
@ -45,6 +43,8 @@ import org.junit.After;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@RunWith(JMock.class) @RunWith(JMock.class)
@ -61,27 +61,34 @@ public class DiscoveryNetworkReconnectTest {
ManagementContext managementContext; ManagementContext managementContext;
DiscoveryAgent agent; DiscoveryAgent agent;
SocketProxy proxy; SocketProxy proxy;
// ignore the hostname resolution component as this is machine dependent // ignore the hostname resolution component as this is machine dependent
class NetworkBridgeObjectNameMatcher<T> extends BaseMatcher<T> { class NetworkBridgeObjectNameMatcher<T> extends BaseMatcher<T> {
T name; T name;
NetworkBridgeObjectNameMatcher(T o) { NetworkBridgeObjectNameMatcher(T o) {
name = o; name = o;
} }
@Override
public boolean matches(Object arg0) { public boolean matches(Object arg0) {
ObjectName other = (ObjectName) arg0; ObjectName other = (ObjectName) arg0;
ObjectName mine = (ObjectName) name; ObjectName mine = (ObjectName) name;
LOG.info("Match: " + mine + " vs: " + other); LOG.info("Match: " + mine + " vs: " + other);
return other.getKeyProperty("Type").equals(mine.getKeyProperty("Type")) &&
other.getKeyProperty("NetworkConnectorName").equals(mine.getKeyProperty("NetworkConnectorName")); if (other.getKeyProperty("service") == null) {
return false;
}
return other.getKeyProperty("service").equals(mine.getKeyProperty("service")) &&
other.getKeyProperty("networkConnectorName").equals(mine.getKeyProperty("networkConnectorName"));
} }
@Override
public void describeTo(Description arg0) { public void describeTo(Description arg0) {
arg0.appendText(this.getClass().getName()); arg0.appendText(this.getClass().getName());
} }
} }
@Before @Before
public void setUp() throws Exception { public void setUp() throws Exception {
context = new JUnit4Mockery() {{ context = new JUnit4Mockery() {{
@ -93,35 +100,38 @@ public class DiscoveryNetworkReconnectTest {
brokerA.addConnector("tcp://localhost:0"); brokerA.addConnector("tcp://localhost:0");
brokerA.start(); brokerA.start();
brokerA.waitUntilStarted(); brokerA.waitUntilStarted();
proxy = new SocketProxy(brokerA.getTransportConnectors().get(0).getConnectUri()); proxy = new SocketProxy(brokerA.getTransportConnectors().get(0).getConnectUri());
managementContext = context.mock(ManagementContext.class); managementContext = context.mock(ManagementContext.class);
context.checking(new Expectations(){{ context.checking(new Expectations(){{
allowing (managementContext).getJmxDomainName(); will (returnValue("Test")); allowing(managementContext).getJmxDomainName(); will (returnValue("Test"));
allowing (managementContext).setBrokerName("BrokerNC"); allowing(managementContext).setBrokerName("BrokerNC");
allowing (managementContext).start(); allowing(managementContext).start();
allowing (managementContext).isCreateConnector(); allowing(managementContext).isCreateConnector();
allowing (managementContext).stop(); allowing(managementContext).stop();
allowing (managementContext).isConnectorStarted(); allowing(managementContext).isConnectorStarted();
// expected MBeans // expected MBeans
allowing (managementContext).registerMBean(with(any(Object.class)), with(equal( allowing(managementContext).registerMBean(with(any(Object.class)), with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Broker")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC"))));
allowing (managementContext).registerMBean(with(any(Object.class)), with(equal( allowing(managementContext).registerMBean(with(any(Object.class)), with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Health")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,service=Health"))));
allowing (managementContext).registerMBean(with(any(Object.class)), with(equal( allowing(managementContext).registerMBean(with(any(Object.class)), with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=NetworkConnector,NetworkConnectorName=NC")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,connector=networkConnectors,networkConnectorName=NC"))));
allowing (managementContext).registerMBean(with(any(Object.class)), with(equal( allowing(managementContext).registerMBean(with(any(Object.class)), with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Topic,Destination=ActiveMQ.Advisory.Connection")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,destinationType=Topic,destinationName=ActiveMQ.Advisory.Connection"))));
allowing (managementContext).registerMBean(with(any(Object.class)), with(equal( allowing(managementContext).registerMBean(with(any(Object.class)), with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Topic,Destination=ActiveMQ.Advisory.NetworkBridge")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,destinationType=Topic,destinationName=ActiveMQ.Advisory.NetworkBridge"))));
allowing (managementContext).registerMBean(with(any(Object.class)), with(equal( allowing(managementContext).registerMBean(with(any(Object.class)), with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=jobScheduler,jobSchedulerName=JMS")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,destinationType=Topic,destinationName=ActiveMQ.Advisory.MasterBroker"))));
allowing(managementContext).registerMBean(with(any(Object.class)), with(equal(
new ObjectName("Test:type=Broker,brokerName=BrokerNC,service=jobScheduler,jobSchedulerName=JMS"))));
atLeast(maxReconnects - 1).of (managementContext).registerMBean(with(any(Object.class)), with(new NetworkBridgeObjectNameMatcher<ObjectName>( atLeast(maxReconnects - 1).of (managementContext).registerMBean(with(any(Object.class)), with(new NetworkBridgeObjectNameMatcher<ObjectName>(
new ObjectName("Test:BrokerName=BrokerNC,Type=NetworkBridge,NetworkConnectorName=NC,Name=localhost/127.0.0.1_" new ObjectName("Test:type=Broker,brokerName=BrokerNC,service=NetworkBridge,networkConnectorName=NC,networkBridgeName=localhost/127.0.0.1_"
+ proxy.getUrl().getPort())))); will(new CustomAction("signal register network mbean") { + proxy.getUrl().getPort())))); will(new CustomAction("signal register network mbean") {
@Override
public Object invoke(Invocation invocation) throws Throwable { public Object invoke(Invocation invocation) throws Throwable {
LOG.info("Mbean Registered: " + invocation.getParameter(0)); LOG.info("Mbean Registered: " + invocation.getParameter(0));
mbeanRegistered.release(); mbeanRegistered.release();
@ -129,27 +139,30 @@ public class DiscoveryNetworkReconnectTest {
} }
}); });
atLeast(maxReconnects - 1).of (managementContext).unregisterMBean(with(new NetworkBridgeObjectNameMatcher<ObjectName>( atLeast(maxReconnects - 1).of (managementContext).unregisterMBean(with(new NetworkBridgeObjectNameMatcher<ObjectName>(
new ObjectName("Test:BrokerName=BrokerNC,Type=NetworkBridge,NetworkConnectorName=NC,Name=localhost/127.0.0.1_" new ObjectName("Test:type=Broker,brokerName=BrokerNC,service=NetworkBridge,networkConnectorName=NC,networkBridgeName=localhost/127.0.0.1_"
+ proxy.getUrl().getPort())))); will(new CustomAction("signal unregister network mbean") { + proxy.getUrl().getPort())))); will(new CustomAction("signal unregister network mbean") {
@Override
public Object invoke(Invocation invocation) throws Throwable { public Object invoke(Invocation invocation) throws Throwable {
LOG.info("Mbean Unregistered: " + invocation.getParameter(0)); LOG.info("Mbean Unregistered: " + invocation.getParameter(0));
mbeanUnregistered.release(); mbeanUnregistered.release();
return null; return null;
} }
}); });
allowing (managementContext).unregisterMBean(with(equal( allowing(managementContext).unregisterMBean(with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Broker")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC"))));
allowing (managementContext).unregisterMBean(with(equal( allowing(managementContext).unregisterMBean(with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Health")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,service=Health"))));
allowing (managementContext).unregisterMBean(with(equal( allowing(managementContext).unregisterMBean(with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=NetworkConnector,NetworkConnectorName=NC")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,connector=networkConnectors,networkConnectorName=NC"))));
allowing (managementContext).unregisterMBean(with(equal( allowing(managementContext).unregisterMBean(with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Topic,Destination=ActiveMQ.Advisory.Connection")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,destinationType=Topic,destinationName=ActiveMQ.Advisory.Connection"))));
allowing (managementContext).unregisterMBean(with(equal( allowing(managementContext).unregisterMBean(with(equal(
new ObjectName("Test:BrokerName=BrokerNC,Type=Topic,Destination=ActiveMQ.Advisory.NetworkBridge")))); new ObjectName("Test:type=Broker,brokerName=BrokerNC,destinationType=Topic,destinationName=ActiveMQ.Advisory.NetworkBridge"))));
allowing(managementContext).unregisterMBean(with(equal(
new ObjectName("Test:type=Broker,brokerName=BrokerNC,destinationType=Topic,destinationName=ActiveMQ.Advisory.MasterBroker"))));
}}); }});
brokerB = new BrokerService(); brokerB = new BrokerService();
brokerB.setManagementContext(managementContext); brokerB.setManagementContext(managementContext);
brokerB.setBrokerName("BrokerNC"); brokerB.setBrokerName("BrokerNC");
@ -164,15 +177,15 @@ public class DiscoveryNetworkReconnectTest {
brokerB.waitUntilStopped(); brokerB.waitUntilStopped();
proxy.close(); proxy.close();
} }
private void configure(BrokerService broker) { private void configure(BrokerService broker) {
broker.setPersistent(false); broker.setPersistent(false);
broker.setUseJmx(true); broker.setUseJmx(true);
} }
@Test @Test
public void testMulicastReconnect() throws Exception { public void testMulicastReconnect() throws Exception {
// control multicast advertise agent to inject proxy // control multicast advertise agent to inject proxy
agent = MulticastDiscoveryAgentFactory.createDiscoveryAgent(new URI(discoveryAddress)); agent = MulticastDiscoveryAgentFactory.createDiscoveryAgent(new URI(discoveryAddress));
agent.registerService(proxy.getUrl().toString()); agent.registerService(proxy.getUrl().toString());
@ -183,42 +196,41 @@ public class DiscoveryNetworkReconnectTest {
brokerB.waitUntilStarted(); brokerB.waitUntilStarted();
doReconnect(); doReconnect();
} }
@Test @Test
public void testSimpleReconnect() throws Exception { public void testSimpleReconnect() throws Exception {
brokerB.addNetworkConnector("simple://(" + proxy.getUrl() brokerB.addNetworkConnector("simple://(" + proxy.getUrl()
+ ")?useExponentialBackOff=false&initialReconnectDelay=500&wireFormat.maxInactivityDuration=1000&wireFormat.maxInactivityDurationInitalDelay=1000"); + ")?useExponentialBackOff=false&initialReconnectDelay=500&wireFormat.maxInactivityDuration=1000&wireFormat.maxInactivityDurationInitalDelay=1000");
brokerB.start(); brokerB.start();
brokerB.waitUntilStarted(); brokerB.waitUntilStarted();
doReconnect(); doReconnect();
} }
private void doReconnect() throws Exception { private void doReconnect() throws Exception {
for (int i=0; i<maxReconnects; i++) { for (int i=0; i<maxReconnects; i++) {
// Wait for connection // Wait for connection
assertTrue("we got a network connection in a timely manner", Wait.waitFor(new Wait.Condition() { assertTrue("we got a network connection in a timely manner", Wait.waitFor(new Wait.Condition() {
@Override
public boolean isSatisified() throws Exception { public boolean isSatisified() throws Exception {
return proxy.connections.size() >= 1; return proxy.connections.size() >= 1;
} }
})); }));
// wait for network connector // wait for network connector
assertTrue("network connector mbean registered within 1 minute", mbeanRegistered.tryAcquire(60, TimeUnit.SECONDS)); assertTrue("network connector mbean registered within 1 minute", mbeanRegistered.tryAcquire(60, TimeUnit.SECONDS));
// force an inactivity timeout via the proxy // force an inactivity timeout via the proxy
proxy.pause(); proxy.pause();
// wait for the inactivity timeout and network shutdown // wait for the inactivity timeout and network shutdown
assertTrue("network connector mbean unregistered within 1 minute", mbeanUnregistered.tryAcquire(60, TimeUnit.SECONDS)); assertTrue("network connector mbean unregistered within 1 minute", mbeanUnregistered.tryAcquire(60, TimeUnit.SECONDS));
// whack all connections // whack all connections
proxy.close(); proxy.close();
// let a reconnect succeed // let a reconnect succeed
proxy.reopen(); proxy.reopen();
} }
} }
} }