Replicator { public void replicate() { try { replicateLoop(); } finally { changeSink.close(); } } Replicator(ReplicationSource source, ChangeSink changeSink,
			TransactionManager snapshotLoader, SystemTimeLoader systemTimeLoader, int iterations,
			int minInterval, int maxInterval); void replicate();  }
@Test public void testInitialization() { Replicator replicator; MockReplicationSource source; MockReplicationDestination destination; MockTransactionSnapshotLoader snapshotLoader; MockSystemTimeLoader timeLoader; ReplicationState state; source = new MockReplicationSource(); destination = new MockReplicationDestination(); snapshotLoader = new MockTransactionSnapshotLoader(); timeLoader = new MockSystemTimeLoader(); replicator = new Replicator(source, destination, snapshotLoader, timeLoader, 1, 0, 0); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:14")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:14")); snapshotLoader.getSnapshots().add(new TransactionSnapshot("100:200:110,112")); replicator.replicate(); state = destination.getCurrentState(); Assert.assertEquals("Incorrect final state.", new ReplicationState( 200, 200, Arrays.asList(new Long[]{110L, 112L}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:14"), 0), state); }
@Test public void testNoAction() { Replicator replicator; MockReplicationSource source; MockReplicationDestination destination; MockTransactionSnapshotLoader snapshotLoader; MockSystemTimeLoader timeLoader; ReplicationState initialState; ReplicationState finalState; initialState = new ReplicationState( 200, 200, Arrays.asList(new Long[]{110L, 112L}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:14"), 0); source = new MockReplicationSource(); destination = new MockReplicationDestination(initialState); snapshotLoader = new MockTransactionSnapshotLoader(); timeLoader = new MockSystemTimeLoader(); replicator = new Replicator(source, destination, snapshotLoader, timeLoader, 1, 0, 0); snapshotLoader.getSnapshots().add(new TransactionSnapshot("100:200:110,112")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); replicator.replicate(); finalState = destination.getCurrentState(); Assert.assertFalse("Final state should not match initial state.", finalState.equals(initialState)); finalState.setTimestamp(initialState.getTimestamp()); finalState.setSequenceNumber(finalState.getSequenceNumber() - 1); Assert.assertTrue("Final state should match initial state after updating timestamp.", finalState.equals(initialState)); Assert.assertTrue("No changes should have been replicated.", source.getPredicatesList().size() == 0); }
@Test public void testSimpleIncrement() { Replicator replicator; MockReplicationSource source; MockReplicationDestination destination; MockTransactionSnapshotLoader snapshotLoader; MockSystemTimeLoader timeLoader; ReplicationState state; ReplicationQueryPredicates predicates; state = new ReplicationState( 200, 200, Arrays.asList(new Long[]{}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:14"), 0); source = new MockReplicationSource(); destination = new MockReplicationDestination(state); snapshotLoader = new MockTransactionSnapshotLoader(); timeLoader = new MockSystemTimeLoader(); replicator = new Replicator(source, destination, snapshotLoader, timeLoader, 1, 0, 0); snapshotLoader.getSnapshots().add(new TransactionSnapshot("100:220")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); replicator.replicate(); state = destination.getCurrentState(); Assert.assertEquals("Incorrect final state.", new ReplicationState( 220, 220, Arrays.asList(new Long[]{}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:15"), 1), state); Assert.assertTrue("A single interval should have been replicated.", source.getPredicatesList().size() == 1); predicates = source.getPredicatesList().get(0); Assert.assertEquals("Incorrect active list.", Collections.emptyList(), predicates.getActiveList()); Assert.assertEquals("Incorrect ready list.", Collections.emptyList(), predicates.getReadyList()); Assert.assertEquals("Incorrect bottom transaction id.", 200, predicates.getBottomTransactionId()); Assert.assertEquals("Incorrect top transaction id.", 220, predicates.getTopTransactionId()); }
@Test public void testInFlightTxnIncrement() { Replicator replicator; MockReplicationSource source; MockReplicationDestination destination; MockTransactionSnapshotLoader snapshotLoader; MockSystemTimeLoader timeLoader; ReplicationState state; ReplicationQueryPredicates predicates; state = new ReplicationState( 200, 200, Arrays.asList(new Long[]{180L, 185L}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:14"), 0); source = new MockReplicationSource(); destination = new MockReplicationDestination(state); snapshotLoader = new MockTransactionSnapshotLoader(); timeLoader = new MockSystemTimeLoader(); replicator = new Replicator(source, destination, snapshotLoader, timeLoader, 1, 0, 0); snapshotLoader.getSnapshots().add(new TransactionSnapshot("100:220:185")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); replicator.replicate(); state = destination.getCurrentState(); Assert.assertEquals("Incorrect final state.", new ReplicationState( 220, 220, Arrays.asList(new Long[]{185L}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:15"), 1), state); Assert.assertTrue("A single interval should have been replicated.", source.getPredicatesList().size() == 1); predicates = source.getPredicatesList().get(0); Assert.assertEquals("Incorrect active list.", Arrays.asList(new Long[]{185L}), predicates.getActiveList()); Assert.assertEquals("Incorrect ready list.", Arrays.asList(new Long[]{180L}), predicates.getReadyList()); Assert.assertEquals("Incorrect bottom transaction id.", 200, predicates.getBottomTransactionId()); Assert.assertEquals("Incorrect top transaction id.", 220, predicates.getTopTransactionId()); }
@Test public void testOutageCatchupWithActiveTxns() { Replicator replicator; MockReplicationSource source; MockReplicationDestination destination; MockTransactionSnapshotLoader snapshotLoader; MockSystemTimeLoader timeLoader; ReplicationState state; ReplicationQueryPredicates predicates; state = new ReplicationState( 5, 5, Arrays.asList(new Long[]{24000L, 26000L}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:14"), 0); source = new MockReplicationSource(); destination = new MockReplicationDestination(state); snapshotLoader = new MockTransactionSnapshotLoader(); timeLoader = new MockSystemTimeLoader(); replicator = new Replicator(source, destination, snapshotLoader, timeLoader, 1, 0, 0); snapshotLoader.getSnapshots().add(new TransactionSnapshot("20000:30000:26000")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); timeLoader.getTimes().add(buildDate("2009-10-11 12:13:15")); replicator.replicate(); state = destination.getCurrentState(); Assert.assertEquals("Incorrect final state.", new ReplicationState( 30000, 25005, Arrays.asList(new Long[]{26000L}), Arrays.asList(new Long[]{}), buildDate("2009-10-11 12:13:14"), 1), state); Assert.assertTrue("A single interval should have been replicated.", source.getPredicatesList().size() == 1); predicates = source.getPredicatesList().get(0); Assert.assertEquals("Incorrect active list.", Arrays.asList(new Long[]{26000L}), predicates.getActiveList()); Assert.assertEquals("Incorrect ready list.", Arrays.asList(new Long[]{}), predicates.getReadyList()); Assert.assertEquals("Incorrect bottom transaction id.", 5, predicates.getBottomTransactionId()); Assert.assertEquals("Incorrect top transaction id.", 25005, predicates.getTopTransactionId()); }