log.info("starting relay on port " + relayPort);
r1.start();
//TestUtil.sleep(10*1000);
// wait until relay comes up
TestUtil.assertWithBackoff(new ConditionCheck() {
@Override
public boolean check() {
return relay1.isRunningStatus();
}
},"Relay hasn't come up completely ", 30000, LOG);
log.info("now create client");
String srcSubscriptionString = TestUtil.join(srcs, ",");
String serverName = "localhost:" + relayPort;
final EventsCountingConsumer countingConsumer = new EventsCountingConsumer();
int id = (RngUtils.randomPositiveInt() % 10000) + 1;
DatabusSourcesConnection clientConn = RelayEventProducer
.createDatabusSourcesConnection("testProducer", id, serverName,
srcSubscriptionString, countingConsumer,
1 * 1024 * 1024, 50000, 30 * 1000, 100, 15 * 1000,
1, true, DatabusClientNettyThreadPools.createNettyThreadPools(id),
0, DbusEventFactory.DBUS_EVENT_V1,0);
cr = new ClientRunner(clientConn);
log.info("starting client");
cr.start();
// wait till client gets the event
TestUtil.assertWithBackoff(new ConditionCheck() {
@Override
public boolean check() {
int events = countingConsumer.getNumDataEvents();
LOG.info("client got " + events + " events");
return events == 2;
}
},"Consumer didn't get 2 events ", 64 * 1024, LOG);
// asserts
Assert.assertEquals(countingConsumer.getNumDataEvents(), 2);
Assert.assertEquals(countingConsumer.getNumWindows(), 2);
Assert.assertEquals(countingConsumer.getNumDataEvents(DbusEventFactory.DBUS_EVENT_V1),2);
log.info("shutdown first client");
clientConn.stop();
cr.shutdown();
TestUtil.sleep(1000);
cr = null;
log.info("start another client who understands V2");
final EventsCountingConsumer countingConsumer1 = new EventsCountingConsumer();
clientConn = RelayEventProducer
.createDatabusSourcesConnection("testProducer", id, serverName,
srcSubscriptionString, countingConsumer1,
1 * 1024 * 1024, 50000, 30 * 1000, 100, 15 * 1000,
1, true, DatabusClientNettyThreadPools.createNettyThreadPools(id),
0, DbusEventFactory.DBUS_EVENT_V2,0);
cr = new ClientRunner(clientConn);
cr.start();
log.info("wait till client gets the event");
TestUtil.assertWithBackoff(new ConditionCheck() {
@Override
public boolean check() {
int events = countingConsumer1.getNumDataEvents();
LOG.debug("client got " + events + " events");
return events == 2;