Async collectors

[jroper]

Currently we're using the JDK8 collector API for doing all terminal accumulations. It would be good to have an async alternative, eg AsyncCollector, where the accumulator and finisher functions return CompletionStage of a value when they are done. Then async alternatives of collect, reduce etc can be provided.

[hutchig]

cf https://stackoverflow.com/questions/17960656/is-it-possible-to-use-java-8-streams-api-for-asynchronous-processing/18826615

[jroper]

Perhaps the simplest async collector is a forEachAsync, which would allow this:

CompletionStage<Void> saveObjectToDatabase(MyObject obj) {
  ...
}

ReactiveStreams.fromPublisher(somePublisher)
  .forEachAsync(this::saveObjectToDatabase)

Currently, the only way to do this is:

ReactiveStreams.fromPublisher(somePublisher)
  .flatMapCompletionStage(this::saveObjectToDatabase)
  .ignore()

The ignore step is counter intuitive here, you're not really ignoring anything, it's just that the only way to do an asynchronous action on each element (with backpressure) currently is to do a flatMapCompletionStage, and then since your database operation returns Void, you don't care about the result, so you ignore it.

The biggest use case though would be if you wanted to do some asynchronous operation, and the next operation depended on the result of the previous. Though, we could do that with an asynchronous version of a fold processing stage in #66, so maybe that's the best place to handle it.