GoldenGateEventProducer extends AbstractEventProducer { public GGParserStatistics getParserStats() { return _ggParserStats; } GoldenGateEventProducer(PhysicalSourceStaticConfig pConfig,
                                 SchemaRegistryService schemaRegistryService,
                                 DbusEventBufferAppendable dbusEventBuffer,
                                 DbusEventsStatisticsCollector statsCollector,
                                 MaxSCNReaderWriter maxScnReaderWriters); GGMonitoredSourceInfo buildGGMonitoredSourceInfo(
            LogicalSourceStaticConfig sourceConfig, PhysicalSourceStaticConfig pConfig); GGParserStatistics getParserStats(); @Override String getName(); @Override long getSCN(); @Override synchronized void start(long sinceSCN); @Override boolean isRunning(); @Override boolean isPaused(); @Override void unpause(); @Override void pause(); @Override synchronized void shutdown(); @Override synchronized void waitForShutdown(); @Override synchronized void waitForShutdown(long timeout); @Override List<? extends EventSourceStatisticsIface> getSources(); GGMonitoredSourceInfo getSource(short sourceId); static final short GLOBAL_SOURCE_ID; final Logger _log; }
@Test public void testSCNRegressionStats() throws Exception { short[] sourceIds = new short[] { 505, 506 }; String[] sourceNames = new String[] { "source1", "source2" }; PhysicalSourceStaticConfig pssc = buildSimplePssc(sourceIds, sourceNames, "gg: DbusEventBufferAppendable mb = createBufMult(pssc); GoldenGateEventProducer gg = new GoldenGateEventProducer(pssc, null, mb, null, null); Object handleXmlCallbackObject = getHandleXmlCallbackInnerInstance(gg); Method method = getOnTransactionEndMethod(); List<String> keys = new ArrayList<String>(); keys.add("key1"); GGParserStatistics ggParserStats = gg.getParserStats(); long timestamp = System.currentTimeMillis() * DbusConstants.NUM_NSECS_IN_MSEC; List<TransactionState.PerSourceTransactionalUpdate> dbUpdates1 = generateUpdates(sourceIds, keys, 10); method.invoke(handleXmlCallbackObject, new Object[] { dbUpdates1, new TransactionInfo(0, 0, timestamp, 10) }); timestamp = System.currentTimeMillis() * DbusConstants.NUM_NSECS_IN_MSEC + 1; Assert.assertEquals("NumSCNRegressions", 0, ggParserStats.getNumSCNRegressions()); Assert.assertEquals("NumSCNRegressions", -1, ggParserStats.getLastRegressedScn()); Assert.assertEquals("MaxScn", 10, ggParserStats.getMaxScn()); List<TransactionState.PerSourceTransactionalUpdate> dbUpdates2 = generateUpdates(sourceIds, keys, 5); method.invoke(handleXmlCallbackObject, new Object[] { dbUpdates2, new TransactionInfo(0, 0, timestamp + 1, 5) }); Assert.assertEquals("NumSCNRegressions", 1, ggParserStats.getNumSCNRegressions()); Assert.assertEquals("NumSCNRegressions", 5, ggParserStats.getLastRegressedScn()); Assert.assertEquals("MaxScn", 10, ggParserStats.getMaxScn()); dbUpdates2 = generateUpdates(sourceIds, keys, 6); method.invoke(handleXmlCallbackObject, new Object[] { dbUpdates2, new TransactionInfo(0, 0, timestamp + 1, 6) }); Assert.assertEquals("NumSCNRegressions", 1, ggParserStats.getNumSCNRegressions()); Assert.assertEquals("NumSCNRegressions", 5, ggParserStats.getLastRegressedScn()); Assert.assertEquals("MaxScn", 10, ggParserStats.getMaxScn()); dbUpdates2 = generateUpdates(sourceIds, keys, 3); method.invoke(handleXmlCallbackObject, new Object[] { dbUpdates2, new TransactionInfo(0, 0, timestamp + 1, 3) }); Assert.assertEquals("NumSCNRegressions", 2, ggParserStats.getNumSCNRegressions()); Assert.assertEquals("NumSCNRegressions", 3, ggParserStats.getLastRegressedScn()); Assert.assertEquals("MaxScn", 10, ggParserStats.getMaxScn()); dbUpdates2 = generateUpdates(sourceIds, keys, 11); method.invoke(handleXmlCallbackObject, new Object[] { dbUpdates2, new TransactionInfo(0, 0, timestamp + 1, 11) }); Assert.assertEquals("NumSCNRegressions", 2, ggParserStats.getNumSCNRegressions()); Assert.assertEquals("NumSCNRegressions", 3, ggParserStats.getLastRegressedScn()); Assert.assertEquals("MaxScn", 11, ggParserStats.getMaxScn()); }