ChangeDeriver implements MultiSinkRunnableChangeSource { public void run() { try { EntityContainerComparator comparator; EntityContainer fromEntityContainer = null; EntityContainer toEntityContainer = null; TimestampSetter timestampSetter; comparator = new EntityContainerComparator(new EntityByTypeThenIdComparator()); timestampSetter = new TimestampSetter(); fromPostbox.outputInitialize(); toPostbox.outputInitialize(); changeSink.initialize(Collections.<String, Object>emptyMap()); while ( (fromEntityContainer != null || fromPostbox.hasNext()) && (toEntityContainer != null || toPostbox.hasNext())) { int comparisonResult; if (fromEntityContainer == null) { fromEntityContainer = fromPostbox.getNext(); } if (toEntityContainer == null) { toEntityContainer = toPostbox.getNext(); } comparisonResult = comparator.compare(fromEntityContainer, toEntityContainer); if (comparisonResult < 0) { changeSink.process( new ChangeContainer( timestampSetter.updateTimestamp(fromEntityContainer), ChangeAction.Delete)); fromEntityContainer = null; } else if (comparisonResult > 0) { changeSink.process(new ChangeContainer(toEntityContainer, ChangeAction.Create)); toEntityContainer = null; } else { if (!fromEntityContainer.getEntity().equals(toEntityContainer.getEntity())) { changeSink.process(new ChangeContainer(toEntityContainer, ChangeAction.Modify)); } fromEntityContainer = null; toEntityContainer = null; } } while (fromEntityContainer != null || fromPostbox.hasNext()) { if (fromEntityContainer == null) { fromEntityContainer = fromPostbox.getNext(); } changeSink.process( new ChangeContainer( timestampSetter.updateTimestamp(fromEntityContainer), ChangeAction.Delete)); fromEntityContainer = null; } while (toEntityContainer != null || toPostbox.hasNext()) { if (toEntityContainer == null) { toEntityContainer = toPostbox.getNext(); } changeSink.process(new ChangeContainer(toEntityContainer, ChangeAction.Create)); toEntityContainer = null; } changeSink.complete(); fromPostbox.outputComplete(); toPostbox.outputComplete(); } finally { changeSink.close(); fromPostbox.outputRelease(); toPostbox.outputRelease(); } } ChangeDeriver(int inputBufferCapacity); Sink getSink(int instance); int getSinkCount(); void setChangeSink(ChangeSink changeSink); void run();  }
@Test public void rightEmpty() throws Exception { ChangeDeriver deriver = new ChangeDeriver(1); RunnableSource left = new XmlReader(dataUtils.createDataFile( "v0_6/derive_change/simple.osm"), true, CompressionMethod.None); RunnableSource right = new EmptyReader(); SinkChangeInspector result = RunTaskUtilities.run(deriver, left, right); List<ChangeContainer> changes = result.getProcessedChanges(); Assert.assertEquals(3, changes.size()); for (ChangeContainer changeContainer : changes) { Assert.assertEquals(ChangeAction.Delete, changeContainer.getAction()); } Entity e; e = changes.get(0).getEntityContainer().getEntity(); Assert.assertEquals(EntityType.Node, e.getType()); Assert.assertEquals(10, e.getId()); Assert.assertEquals(34, e.getVersion()); e = changes.get(1).getEntityContainer().getEntity(); Assert.assertEquals(EntityType.Way, e.getType()); Assert.assertEquals(100, e.getId()); Assert.assertEquals(56, e.getVersion()); e = changes.get(2).getEntityContainer().getEntity(); Assert.assertEquals(EntityType.Relation, e.getType()); Assert.assertEquals(1000, e.getId()); Assert.assertEquals(78, e.getVersion()); }