CustomMapAggregator extends KorypheBinaryOperator<CustomMap<K, V>> { public void setBinaryOperator(final BinaryOperator<V> binaryOperator) { this.binaryOperator = binaryOperator; } CustomMapAggregator();  CustomMapAggregator(final BinaryOperator<V> binaryOperator); void setBinaryOperator(final BinaryOperator<V> binaryOperator); BinaryOperator<? super V> getBinaryOperator(); CustomMap<K, V> _apply(final CustomMap<K, V> state, final CustomMap<K, V> input);  }
@Test public void testMapAggregation() { int inA = 1; int inB = 2; int noInputs = 3; CustomMap<String, Integer>[] inputs = new CustomMap[noInputs]; for (int i = 0; i < noInputs; i++) { inputs[i] = new CustomMap<>(new StringSerialiser(), new IntegerSerialiser()); inputs[i].put("a", inA); inputs[i].put("b", inB); } BinaryOperator<Integer> aggregator = mock(BinaryOperator.class); for (int i = 0; i < noInputs; i++) { Integer expectedA = null; Integer expectedB = null; if (i > 0) { expectedA = i * inA; expectedB = i * inB; } given(aggregator.apply(expectedA, inA)).willReturn(inA + (expectedA == null ? 0 : expectedA)); given(aggregator.apply(expectedB, inB)).willReturn(inB + (expectedB == null ? 0 : expectedB)); } CustomMapAggregator<String, Integer> mapBinaryOperator = new CustomMapAggregator<>(); mapBinaryOperator.setBinaryOperator(aggregator); CustomMap<String, Integer> state = null; for (CustomMap<String, Integer> input : inputs) { state = mapBinaryOperator.apply(state, input); } assertEquals(noInputs * inA, (int) state.get("a")); assertEquals(noInputs * inB, (int) state.get("b")); }