Scatter-Gather Using Spring Reactor Core

I have had a good working experience using the Netflix Rx-Java libraries and have previously blogged about using Rx-Java and Java 8 CompletableFuture for scatter-gather kind of problems. Here I want to explore applying the same pattern using the Spring Reactor Core library.

If you are familiar with Netflix Rx-Java, you already know Spring Reactor Core. The API's map beautifully, and I was thrilled to see that the Spring Reactor team has diligently used Marble diagrams in their Javadoc API's

Another quick point is that rx.Observable maps to Flux or Mono based on whether many items are being emitted or whether one or none is being emitted.

With this, let me directly jump into the sample. I have a simple task (simulated using a delay) that is spawned a few times. I need to execute this task multiple times concurrently and then collect back the results, represented the following way using an rx.Observable:

@Test
public void testScatterGather() throws Exception {
    ExecutorService executors = Executors.newFixedThreadPool(5);

    List<Observable<String>> obs =
            IntStream.range(0, 10)
                .boxed()
                .map(i -> generateTask(i, executors)).collect(Collectors.toList());


    Observable<List<String>> merged = Observable.merge(obs).toList();
    List<String> result = merged.toBlocking().first();

    logger.info(result.toString());

}

private Observable<String> generateTask(int i, ExecutorService executorService) {
    return Observable
            .<String>create(s -> {
                Util.delay(2000);
                s.onNext( i + "-test");
                s.onCompleted();
            }).subscribeOn(Schedulers.from(executorService));
}

Note that I am blocking purely for the test.

Now, similar code using Spring Reactor Core:

@Test
public void testScatterGather() {
    ExecutorService executors = Executors.newFixedThreadPool(5);

    List<Flux<String>> fluxList = IntStream.range(0, 10)
            .boxed()
            .map(i -> generateTask(executors, i)).collect(Collectors.toList());

    Mono<List<String>> merged = Flux.merge(fluxList).toList();

    List<String> list = merged.get();

    logger.info(list.toString());


}

public Flux<String> generateTask(ExecutorService executorService, int i) {
    return Flux.<String>create(s -> {
        Util.delay(2000);
        s.onNext(i + "-test");
        s.onComplete();
    }).subscribeOn(executorService);
}

It more or less maps one to one. A small difference is in the Mono type. I personally felt that this type was a nice introduction to the reactive library as it makes it very clear whether more than 1 item is being emitted vs. only a single item, as in the sample.

These are still early explorations for me and I look forward to getting far more familiar with this excellent library.