ItemMapper { public Item mergeItem(PrimaryKey primaryKey, Item newItem, Item oldItem) { Item resultItem = new Item().withPrimaryKey(primaryKey); Set<String> visited = new HashSet<>(); visited.add(CAMPAIGN_TABLE_HASH_KEY); visited.add(PERIOD_TABLE_RANGE_KEY); mergeOneWay(newItem, oldItem, resultItem, visited); mergeOneWay(oldItem, newItem, resultItem, visited); return resultItem; }  PrimaryKey key(ByteBuffer byteBuffer); PrimaryKey key(JsonNode node); PrimaryKey primaryKey(Item item); Map<PrimaryKey, Item> mergeItems(List<ByteBuffer> list); List<Item> mergeItems(Collection<Item> created, Collection<Item> read); Item mergeItem(PrimaryKey primaryKey, Item newItem, Item oldItem); Object mergeValue(Object newObject, Object oldValue); Item fromByteBuffer(ByteBuffer byteBuffer); static final String CAMPAIGN_TABLE_HASH_KEY; static final String PERIOD_TABLE_RANGE_KEY; }
@Test public void mergeItem() throws Exception { Aggregation aggregation1 = aggregations.get(0); Aggregation aggregation2 = aggregations.get(1); Item aggregation1_item = itemMapper.fromByteBuffer(serialize(aggregation1)); Item aggregation2_item = itemMapper.fromByteBuffer(serialize(aggregation2)); Item resultItem = itemMapper.mergeItem(itemMapper.primaryKey(aggregation1_item), aggregation1_item, aggregation2_item); assertEquals(itemMapper.primaryKey(aggregation2_item), itemMapper.primaryKey(resultItem)); assertEquals(aggregation1.getClicks() + aggregation2.getClicks(), resultItem.getLong("clicks")); assertEquals(aggregation1.getBids() + aggregation2.getBids(), resultItem.getLong("bids")); assertEquals(aggregation1.getImps() + aggregation2.getImps(), resultItem.getLong("imps")); }