int relayPort = Utils.getAvailablePort(11993);
// create relay
final DatabusRelayMain relay1 = createRelay(relayPort, pId, srcs);
DatabusRelayTestUtil.RelayRunner r1=null;
ClientRunner cr = null;
try
{
//EventProducer[] producers = relay1.getProducers();
r1 = new DatabusRelayTestUtil.RelayRunner(relay1);
log.info("Relay created");
DbusEventBufferMult bufMult = relay1.getEventBuffer();
PhysicalPartition pPartition = new PhysicalPartition((int)pId, pSourceName);
DbusEventBuffer buf = (DbusEventBuffer)bufMult.getDbusEventBufferAppendable(pPartition);
log.info("create some events");
long windowScn = 100L;
ByteBuffer serializationBuffer = addEvent(windowScn, srcId, relay1.getSchemaRegistryService().fetchSchemaIdForSourceNameAndVersion(srcs[0], 2).getByteArray(),
pId, DbusEventFactory.DBUS_EVENT_V2);
ReadableByteChannel channel = Channels.newChannel(new ByteBufferInputStream(serializationBuffer));
int readEvents = buf.readEvents(channel);
log.info("successfully read in " + readEvents + " events ");
channel.close();
windowScn = 101L;
serializationBuffer = addEvent(windowScn, srcId, relay1.getSchemaRegistryService().fetchSchemaIdForSourceNameAndVersion(srcs[0], 2).getByteArray(),
pId, DbusEventFactory.DBUS_EVENT_V1);
channel = Channels.newChannel(new ByteBufferInputStream(serializationBuffer));
readEvents = buf.readEvents(channel);
log.info("successfully read in " + readEvents + " events ");
channel.close();
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();