GoldenGateEventProducer extends AbstractEventProducer { @Override public long getSCN() { return _scn.get(); } 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 testDbUpdatesMergeUtils() throws Exception { short[] sourceIds = new short[] { 505, 506 }; long scn = 10; { String key = "key1"; String oldVal = "val1"; Map<String, String> keyVals1 = new HashMap<String, String>(); keyVals1.put(key, oldVal); Map<String, String> keyVals2 = new HashMap<String, String>(); String newVal = "val2"; keyVals2.put(key, newVal); List<TransactionState.PerSourceTransactionalUpdate> oldDbUpdates = generateUpdatesForSchema2(sourceIds, keyVals1, scn); List<TransactionState.PerSourceTransactionalUpdate> newDbUpdates = generateUpdatesForSchema2(sourceIds, keyVals2, scn); List<TransactionState.PerSourceTransactionalUpdate> result = DBUpdatesMergeUtils.mergeTransactionData(newDbUpdates, oldDbUpdates); Assert.assertEquals("Number of Source batches", 2, result.size()); Assert.assertEquals("Result Set Size", result.get(0).getDbUpdatesSet().size(), 1); Assert.assertEquals("Result Set Size", result.get(1).getDbUpdatesSet().size(), 1); DBUpdateImage dbi = result.get(0).getDbUpdatesSet().iterator().next(); DBUpdateImage expDbi = newDbUpdates.get(0).getDbUpdatesSet().iterator().next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); String k = (String) dbi.getGenericRecord().get("name1"); String v = (String) dbi.getGenericRecord().get("name2"); String expK = (String) expDbi.getGenericRecord().get("name1"); String expV = (String) expDbi.getGenericRecord().get("name2"); Assert.assertEquals("Key Check", expK, k); Assert.assertEquals("Val Check", expV, v); dbi = result.get(1).getDbUpdatesSet().iterator().next(); expDbi = newDbUpdates.get(1).getDbUpdatesSet().iterator().next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); k = (String) dbi.getGenericRecord().get("name1"); v = (String) dbi.getGenericRecord().get("name2"); expK = (String) expDbi.getGenericRecord().get("name1"); expV = (String) expDbi.getGenericRecord().get("name2"); Assert.assertEquals("Key Check", expK, k); Assert.assertEquals("Val Check", expV, v); } { String key = "key1"; String oldVal = "val1"; Map<String, String> keyVals1 = new HashMap<String, String>(); keyVals1.put(key, oldVal); Map<String, String> keyVals2 = new HashMap<String, String>(); key = "key2"; String newVal = "val2"; keyVals2.put(key, newVal); List<TransactionState.PerSourceTransactionalUpdate> oldDbUpdates = generateUpdatesForSchema2(sourceIds, keyVals1, scn); List<TransactionState.PerSourceTransactionalUpdate> newDbUpdates = generateUpdatesForSchema2(sourceIds, keyVals2, scn); List<TransactionState.PerSourceTransactionalUpdate> result = DBUpdatesMergeUtils.mergeTransactionData(newDbUpdates, oldDbUpdates); Assert.assertEquals("Number of Source batches", 2, result.size()); Assert.assertEquals("Result Set Size", result.get(0).getDbUpdatesSet().size(), 2); Assert.assertEquals("Result Set Size", result.get(1).getDbUpdatesSet().size(), 2); Iterator<DBUpdateImage> itr = result.get(0).getDbUpdatesSet().iterator(); DBUpdateImage dbi = itr.next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); String k = (String) dbi.getGenericRecord().get("name1"); String v = (String) dbi.getGenericRecord().get("name2"); dbi = itr.next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); String k2 = (String) dbi.getGenericRecord().get("name1"); String v2 = (String) dbi.getGenericRecord().get("name2"); if (keyVals1.containsKey(k)) { Assert.assertEquals("Val Check", keyVals1.get(k), v); Assert.assertEquals("Val Check 2", keyVals2.get(k2), v2); } else { Assert.assertEquals("Val Check", keyVals2.get(k), v); Assert.assertEquals("Val Check 2", keyVals1.get(k2), v2); } itr = result.get(1).getDbUpdatesSet().iterator(); dbi = itr.next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); k = (String) dbi.getGenericRecord().get("name1"); v = (String) dbi.getGenericRecord().get("name2"); dbi = itr.next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); k2 = (String) dbi.getGenericRecord().get("name1"); v2 = (String) dbi.getGenericRecord().get("name2"); if (keyVals1.containsKey(k)) { Assert.assertEquals("Val Check", keyVals1.get(k), v); Assert.assertEquals("Val Check 2", keyVals2.get(k2), v2); } else { Assert.assertEquals("Val Check", keyVals2.get(k), v); Assert.assertEquals("Val Check 2", keyVals1.get(k2), v2); } } { String key = "key1"; String oldVal = "val1"; Map<String, String> keyVals1 = new HashMap<String, String>(); keyVals1.put(key, oldVal); Map<String, String> keyVals2 = new HashMap<String, String>(); key = "key2"; String newVal = "val2"; keyVals2.put(key, newVal); List<TransactionState.PerSourceTransactionalUpdate> oldDbUpdates = generateUpdatesForSchema2(new short[] { sourceIds[0] }, keyVals1, scn); List<TransactionState.PerSourceTransactionalUpdate> newDbUpdates = generateUpdatesForSchema2(new short[] { sourceIds[1] }, keyVals2, scn); List<TransactionState.PerSourceTransactionalUpdate> result = DBUpdatesMergeUtils.mergeTransactionData(newDbUpdates, oldDbUpdates); Assert.assertEquals("Number of Source batches", 2, result.size()); Assert.assertEquals("Result Set Size", result.get(0).getDbUpdatesSet().size(), 1); Assert.assertEquals("Result Set Size", result.get(1).getDbUpdatesSet().size(), 1); Iterator<DBUpdateImage> itr = result.get(0).getDbUpdatesSet().iterator(); DBUpdateImage dbi = itr.next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); String k = (String) dbi.getGenericRecord().get("name1"); String v = (String) dbi.getGenericRecord().get("name2"); Assert.assertEquals("Val Check", keyVals1.get(k), v); itr = result.get(1).getDbUpdatesSet().iterator(); dbi = itr.next(); Assert.assertEquals("SCN Check", scn, dbi.getScn()); k = (String) dbi.getGenericRecord().get("name1"); v = (String) dbi.getGenericRecord().get("name2"); Assert.assertEquals("Val Check", keyVals2.get(k), v); } }