How do you execute map reduce operations with the Reactor framework?

Question

Can someone please show a code example of how to perform map/reduce operations via the Reactor framework?

Let's say I have a Collection<Map>. I want to:

1. Transform each Map instance to an object of type Foo concurrently (each instance is totally independent of another - there is no need to convert each serially/iteratively).

2. When all of them are converted, I want a a method, onReduce(Collection<Foo> foos), to be called - the argument contains all of the resulting Foo instances.

Answer 1

Seems to me you don't need reduce at all. The collect and consume are for you:

@Test
public void testCollect() {
    Stream<String> stream = Streams.defer(Arrays.asList("1", "2", "3", "4", "5")).get();
    stream.map(Integer::parseInt)
            .collect()
            .consume(integers -> assertThat(integers, Matchers.contains(1, 2, 3, 4, 5)));
}

This sample (Java 8) demonstrates how to send a List<String> to Reactor's Stream, convert each item to String, collect them to the List<Integer> and do process on the result List.

UPDATE

Note collect(5) isn't needed: Stream applies batchSize from deffered Collection. Looks like .collect(int batchSize) has been introduced recently.