ItemMapper { public Map<PrimaryKey, Item> mergeItems(List<ByteBuffer> list) { return list.stream() .map( r -> new AbstractMap.SimpleEntry<>( key(r), fromByteBuffer(r) ) ).collect( Collectors.toMap( AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue, (left,right) -> mergeItem( primaryKey(left), left, right ) ) ); }  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 mergeItems() { Map<PrimaryKey, Item> merged = itemMapper.mergeItems( aggregations.stream() .map(this::serialize) .collect(Collectors.toList()) ); assertEquals(1, merged.size()); Item mergedItem = merged.values().iterator().next(); Aggregation aggregation1 = aggregations.get(0); Aggregation aggregation2 = aggregations.get(1); assertEquals(aggregation1.getClicks() + aggregation2.getClicks(), mergedItem.getLong("clicks")); assertEquals(aggregation1.getBids() + aggregation2.getBids(), mergedItem.getLong("bids")); assertEquals(aggregation1.getImps() + aggregation2.getImps(), mergedItem.getLong("imps")); }