// actually attempt to start bridges. Post-patch, only one thread is
// allowed to start the bridge.
final CountDownLatch attemptLatch = new CountDownLatch(2);
final CountDownLatch createLatch = new CountDownLatch(2);
DiscoveryNetworkConnector nc = new DiscoveryNetworkConnector() {
@Override
public void onServiceAdd(DiscoveryEvent event) {
// Pre-and-post patch, two threads attempt to establish a bridge
// to the same remote broker.
attemptLatch.countDown();
super.onServiceAdd(event);
}
@Override
protected NetworkBridge createBridge(Transport localTransport,
Transport remoteTransport, final DiscoveryEvent event) {
// Pre-patch, the two threads are allowed to create the bridge.
// Post-patch, only the first thread is allowed. Wait a
// reasonable delay once both attempts are detected to allow
// the two bridge creations to occur concurrently (pre-patch).
// Post-patch, the wait will timeout and allow the first (and
// only) bridge creation to occur.
try {
attemptLatch.await();
createLatch.countDown();
createLatch.await(ATTEMPT_TO_CREATE_DELAY,
TimeUnit.MILLISECONDS);
return super.createBridge(localTransport, remoteTransport,
event);
} catch (InterruptedException e) {
Thread.interrupted();
return null;
}
}
};
nc.setDiscoveryAgent(new DiscoveryAgent() {
TaskRunnerFactory taskRunner = new TaskRunnerFactory();
DiscoveryListener listener;
@Override
public void start() throws Exception {
taskRunner.init();
taskRunner.execute(new Runnable() {
@Override
public void run() {
listener.onServiceAdd(new DiscoveryEvent(broker2
.getVmConnectorURI().toString()));
}
});
taskRunner.execute(new Runnable() {
@Override
public void run() {
listener.onServiceAdd(new DiscoveryEvent(broker2
.getVmConnectorURI().toString()));
}
});
}
@Override
public void stop() throws Exception {
taskRunner.shutdown();
}
@Override
public void setDiscoveryListener(DiscoveryListener listener) {
this.listener = listener;
}
@Override
public void registerService(String name) throws IOException {
}
@Override
public void serviceFailed(DiscoveryEvent event) throws IOException {
listener.onServiceRemove(event);
}
});
broker1.addNetworkConnector(nc);
nc.start();
// Wait for the bridge to be formed by the first attempt.
waitForBridge(broker1.getBrokerName(), broker2.getBrokerName(),
MAX_TEST_TIME, TimeUnit.MILLISECONDS);
// Pre-patch, the second bridge creation attempt fails and removes the
// first (successful) bridge creation attempt from the
// list of active bridges. Post-patch, the second bridge creation
// attempt is prevented, so the first bridge creation attempt
// remains "active". This assertion is expected to fail pre-patch and
// pass post-patch.
Assert.assertFalse(nc.activeBridges().isEmpty());
}