final DatabusRelayMain relay3 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1020,
relayPort, 10 * 1024 * 1024, srcConfigs, SCHEMA_REGISTRY_DIR);
Assert.assertNotNull(relay1);
Assert.assertNotNull(relay3);
r1 = new DatabusRelayTestUtil.RelayRunner(relay1);
final DbusEventsTotalStats stats = relay1.getInboundEventStatisticsCollector().getTotalStats();
final DbusEventsTotalStats stats3 = relay3.getInboundEventStatisticsCollector().getTotalStats();
// create chained relay
PhysicalSourceConfig[] chainedSrcConfigs = new PhysicalSourceConfig[srcNames.length];
int j = 0;
for (String[] srcs : srcNames)
{
PhysicalSourceConfig src1 = DatabusRelayTestUtil.createPhysicalConfigBuilder(
(short) (j + 1),DatabusRelayTestUtil.getPhysicalSrcName(srcs[0]),
"localhost:" + relayPort, 500, eventRatePerSec,0,largestEventSize,largestWindowSize,srcs);
chainedSrcConfigs[j++] = src1;
}
int chainedRelayPort = relayPort + 1;
final DatabusRelayMain relay2 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1021,
chainedRelayPort, 10 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR);
Assert.assertNotNull(relay2);
r2 = new DatabusRelayTestUtil.RelayRunner(relay2);
resetSCN(relay2);
// now create client:
String srcSubscriptionString = TestUtil.join(srcNames[0], ",");
String serverName = "localhost:" + chainedRelayPort;
CountingConsumer countingConsumer = new CountingConsumer();
DatabusSourcesConnection clientConn = RelayEventProducer
.createDatabusSourcesConnection("testProducer", serverName,
srcSubscriptionString, countingConsumer,
1 * 1024 * 1024, largestEventSize, 30 * 1000, 100, 15 * 1000,
1, true,largestEventSize/10);
cr = new ClientRunner(clientConn);
// async starts for all components;
r1.start();
Thread.sleep(10*1000);
// start chained relay
r2.start();
//start client
cr.start();
//Pause r1;
r1.pause();
Thread.sleep(1000);
long firstGenDataEvents = stats.getNumDataEvents();
long firstGenMinScn = stats.getMinScn();
long firstGenWindows = stats.getNumSysEvents();
Assert.assertTrue(stats.getNumSysEvents() > 0);
log.warn("relay1: numDataEvents=" + firstGenDataEvents
+ " numWindows=" + firstGenWindows + " minScn="
+ firstGenMinScn + " maxScn=" + stats.getMaxScn());
Thread.sleep(4*1000);
//clear the relay
boolean s = r1.shutdown(2000);
Assert.assertTrue(s);
DbusEventsTotalStats stats2 = relay2.getInboundEventStatisticsCollector().getTotalStats();
long firstGenChainWindows = stats2.getNumSysEvents();
log.warn("relay2: numDataEvents=" + stats2.getNumDataEvents()
+ " numWindows=" + firstGenChainWindows + " minScn="
+ stats2.getMinScn() + " maxScn=" + stats2.getMaxScn());
Thread.sleep(2*1000);
//restart relay
r3 = new DatabusRelayTestUtil.RelayRunner(relay3);
r3.start();
Thread.sleep(15*1000);
r3.pause();
Thread.sleep(35*1000);
log.warn("relay3: numDataEvents=" + stats3.getNumDataEvents()
+ " numWindows=" + stats3.getNumSysEvents() + " minScn="
+ stats3.getMinScn() + " maxScn=" + stats3.getMaxScn());
stats2 = relay2.getInboundEventStatisticsCollector().getTotalStats();
log.warn("relay2b: numDataEvents=" + stats2.getNumDataEvents()
+ " numWindows=" + stats2.getNumSysEvents() + " minScn="
+ stats2.getMinScn() + " maxScn=" + stats2.getMaxScn());
log.warn("consumer: " + countingConsumer);
//compare chained relays with 2 gens of tier 0 relays
Assert.assertEquals(stats2.getMinScn(), firstGenMinScn) ;
Assert.assertEquals(stats2.getMaxScn(), stats3.getMaxScn());
//the total event windows seen by the chained relay will be state of consumption at first failure of relay1 minus 1 overlap window
Assert.assertEquals(stats2.getNumSysEvents(), (firstGenChainWindows-1) + stats3.getNumSysEvents());
Assert.assertTrue(stats2.getNumDataEvents() > stats3.getNumDataEvents());
//compare source to final client
Assert.assertEquals(countingConsumer.getNumSources(), 2);
Assert.assertEquals(stats2.getNumSysEvents(), countingConsumer
.getNumWindows());
boolean sorted= true;
long prev = -1;
log.info(" scn seq on consumer=");