.setProducerBufferSize(numEvents*2)
.setPayloadSize(100)
.setDeleteInterval(1)
.setProdQueuePolicy(QueuePolicy.OVERWRITE_ON_WRITE);
input.setTestName("testStreamScn");
DbusEventsStatisticsCollector emitterStats = new DbusEventsStatisticsCollector(1,"appenderStats",true,true,null);
DbusEventsStatisticsCollector clientStats = new DbusEventsStatisticsCollector(1,"clientStats",true,true,null);
DbusEventGenerator evGen = new DbusEventGenerator();
assertTrue(evGen.generateEvents(numEvents,input.getWindowSize(),512,input.getPayloadSize(),true,srcTestEvents) > 0);
int eventSize = srcTestEvents.get(0).size();
DbusEventBuffer prodEventBuffer =
new DbusEventBuffer(getConfig(input.getProducerBufferSize()*eventSize,
input.getIndividualBufferSize()*eventSize,
input.getIndexSize()*eventSize,
input.getStagingBufferSize()*eventSize,
AllocationPolicy.HEAP_MEMORY, input.getProdQueuePolicy(),
input.getProdBufferAssertLevel()));
DbusEventBuffer readEventBuffer =
new DbusEventBuffer(getConfig(input.getProducerBufferSize()*eventSize,
input.getIndividualBufferSize()*eventSize, input.getIndexSize()*eventSize,
input.getStagingBufferSize()*eventSize,
AllocationPolicy.HEAP_MEMORY, input.getProdQueuePolicy(),
input.getProdBufferAssertLevel()));
DbusEventAppender eventProducer = new DbusEventAppender(srcTestEvents, prodEventBuffer,emitterStats);
Vector<Long> seenScns = new Vector<Long>();
Checkpoint cp = new Checkpoint();
cp.setConsumptionMode(DbusClientMode.ONLINE_CONSUMPTION);
boolean origEmptyValue = prodEventBuffer.empty();
//empty buffer; prevScn=-1 , minScn=-1 ; so no Scn not found exception
cp.setWindowScn(2L);
cp.setWindowOffset(-1);
seenScns.clear();
readEventBuffer.clear();
boolean res = streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertEquals(-1L, prodEventBuffer.getPrevScn());
assertTrue(res);
assertEquals(0, seenScns.size());
//partial buffer; with no complete window written; prevScn > sinceScn , minScn=-1 ; Scn not found exception thrown;
cp.setWindowScn(2L);
cp.setWindowOffset(-1);
prodEventBuffer.setPrevScn(20L);
prodEventBuffer.setEmpty(false);
seenScns.clear();
readEventBuffer.clear();
assertEquals(-1L, prodEventBuffer.getMinScn());
res = streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertFalse(res);
assertEquals(0, seenScns.size());
//restore
prodEventBuffer.setPrevScn(-1L);
prodEventBuffer.setEmpty(origEmptyValue);
//partial buffer; with no complete window written; sinceScn >= prevScn , minScn=-1 ; no exception;
cp.setWindowScn(45L);
cp.setWindowOffset(-1);
prodEventBuffer.setPrevScn(20L);
prodEventBuffer.setEmpty(false);
seenScns.clear();
readEventBuffer.clear();
res = streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertTrue(res);
assertEquals(0, seenScns.size());
//restore
prodEventBuffer.setPrevScn(-1L);
prodEventBuffer.setEmpty(origEmptyValue);
Thread tEmitter = new Thread(eventProducer);
tEmitter.start();
tEmitter.join();
long minScn = emitterStats.getTotalStats().getMinScn();
long maxScn = emitterStats.getTotalStats().getMaxScn();
long prevScn = emitterStats.getTotalStats().getPrevScn();
System.out.printf("minScn=%d,maxScn=%d,prevScn=%d,range=%d\n",minScn,maxScn,prevScn,emitterStats.getTotalStats().getTimeSpan());
assertEquals(numEvents - 1, emitterStats.getTotalStats().getTimeSpan()/MILLISECONDS);
assertEquals(prodEventBuffer.getTimestampOfFirstEvent(), emitterStats.getTotalStats().getTimestampMinScnEvent());
//stream with scn < max; expect last window; not last 2
cp.setWindowScn(maxScn-1);
cp.setWindowOffset(-1);
seenScns.clear();
readEventBuffer.clear();
streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertEquals(1, seenScns.size());
assertEquals(Long.valueOf(maxScn), seenScns.get(0));
//set windowScn to maxScn; get >= behaviour here ; get the last window
cp.setWindowScn(maxScn);
cp.setWindowOffset(0);
seenScns.clear();
readEventBuffer.clear();
streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertEquals(1, seenScns.size());
assertEquals(Long.valueOf(maxScn), seenScns.get(0));
//stream with scn >= max ; get a window higher than max - expect nothing
cp.setWindowScn(maxScn);
cp.setWindowOffset(-1);
readEventBuffer.clear();
seenScns.clear();
streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertEquals(0, seenScns.size());
//stream with scn > max
cp.setWindowScn(maxScn+1);
cp.setWindowOffset(-1);
readEventBuffer.clear();
seenScns.clear();
streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertEquals(0, seenScns.size());
//stream with scn >= min
cp.setWindowScn(minScn);
cp.setWindowOffset(-1);
readEventBuffer.clear();
seenScns.clear();
streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns);
assertEquals(numScns-1, seenScns.size());
assertTrue(seenScns.get(0)!=minScn);
assertEquals(Long.valueOf(maxScn), seenScns.get(numScns-2));
//stream with scn < min but >= prevScn
cp.setWindowScn(prevScn);
cp.setWindowOffset(-1);
readEventBuffer.clear();
seenScns.clear();
streamWriterReader(prodEventBuffer,input.getBatchSize()*eventSize,cp,"scn",readEventBuffer,seenScns,clientStats);
assertEquals(numScns, seenScns.size());
assertEquals(Long.valueOf(minScn), seenScns.get(0));
System.out.printf("Clientstats: minScn=%d , maxScn=%d , timespan=%d timeSinceLastEvent=%d\n",
clientStats.getTotalStats().getMinScn(),
clientStats.getTotalStats().getMaxScn(),
clientStats.getTotalStats().getTimeSpan(),
clientStats.getTotalStats().getTimeSinceLastEvent());
assertEquals(maxScn, clientStats.getTotalStats().getMaxScn());
assertEquals(numEvents - 1, clientStats.getTotalStats().getTimeSpan()/MILLISECONDS);
assertEquals(clientStats.getTotalStats().getTimestampMaxScnEvent(),
emitterStats.getTotalStats().getTimestampMaxScnEvent());
assertEquals(minScn, clientStats.getTotalStats().getMinScn());
//stream with scn < prevScn
cp.setWindowScn(prevScn-1);
cp.setWindowOffset(-1);
readEventBuffer.clear();