AggregationGroupReducer extends AbstractStreamConsumer<T> implements StreamDataAcceptor<T> { public Promise<List<AggregationChunk>> getResult() { return chunksCollector.get(); } AggregationGroupReducer(@NotNull AggregationChunkStorage<C> storage,
			@NotNull AggregationStructure aggregation, @NotNull List<String> measures,
			@NotNull Class<T> recordClass, @NotNull PartitionPredicate<T> partitionPredicate,
			@NotNull Function<T, K> keyFunction, @NotNull Aggregate<T, Object> aggregate,
			int chunkSize, @NotNull DefiningClassLoader classLoader); Promise<List<AggregationChunk>> getResult(); @Override void accept(T item); void flush(); int getBufferSize(); @Override String toString();  }
@Test public void test() { DefiningClassLoader classLoader = DefiningClassLoader.create(); AggregationStructure structure = AggregationStructure.create(ChunkIdCodec.ofLong()) .withKey("word", FieldTypes.ofString()) .withMeasure("documents", union(ofInt())); List<StreamConsumer> listConsumers = new ArrayList<>(); List items = new ArrayList(); AggregationChunkStorage<Long> aggregationChunkStorage = new AggregationChunkStorage<Long>() { long chunkId; @Override public <T> Promise<StreamSupplier<T>> read(AggregationStructure aggregation, List<String> fields, Class<T> recordClass, Long chunkId, DefiningClassLoader classLoader) { return Promise.of(StreamSupplier.ofIterable(items)); } @Override public <T> Promise<StreamConsumer<T>> write(AggregationStructure aggregation, List<String> fields, Class<T> recordClass, Long chunkId, DefiningClassLoader classLoader) { StreamConsumerToList consumer = StreamConsumerToList.create(items); listConsumers.add(consumer); return Promise.of(consumer); } @Override public Promise<Long> createId() { return Promise.of(++chunkId); } @Override public Promise<Void> finish(Set<Long> chunkIds) { return Promise.complete(); } }; Class<InvertedIndexRecord> inputClass = InvertedIndexRecord.class; Class<Comparable> keyClass = createKeyClass( keysToMap(Stream.of("word"), structure.getKeyTypes()::get), classLoader); Class<InvertedIndexRecord> aggregationClass = createRecordClass(structure, asList("word"), asList("documents"), classLoader); Function<InvertedIndexRecord, Comparable> keyFunction = createKeyFunction(inputClass, keyClass, asList("word"), classLoader); Aggregate<InvertedIndexRecord, Object> aggregate = createPreaggregator(structure, inputClass, aggregationClass, singletonMap("word", "word"), singletonMap("documents", "documentId"), classLoader); int aggregationChunkSize = 2; StreamSupplier<InvertedIndexRecord> supplier = StreamSupplier.of( new InvertedIndexRecord("fox", 1), new InvertedIndexRecord("brown", 2), new InvertedIndexRecord("fox", 3), new InvertedIndexRecord("brown", 3), new InvertedIndexRecord("lazy", 4), new InvertedIndexRecord("dog", 1), new InvertedIndexRecord("quick", 1), new InvertedIndexRecord("fox", 4), new InvertedIndexRecord("brown", 10)); AggregationGroupReducer<Long, InvertedIndexRecord, Comparable> groupReducer = new AggregationGroupReducer<>(aggregationChunkStorage, structure, asList("documents"), aggregationClass, singlePartition(), keyFunction, aggregate, aggregationChunkSize, classLoader); await(supplier.streamTo(groupReducer)); List<AggregationChunk> list = await(groupReducer.getResult()); assertEndOfStream(supplier); assertEndOfStream(groupReducer); assertEquals(5, list.size()); for (StreamConsumer consumer : listConsumers) { assertEndOfStream(consumer); } }