out of order processing of a Flux of input using Reactor 0.9.1.RELEASE

BlockingJournal.java

import reactor.core.publisher.Mono;

public class BlockingJournal {

    private static String blockingWrite(String in){
        try {
            // fakes blocking for disk write
            Thread.sleep(5L);
            System.out.println("journal wrote: "+in+" on "+Thread.currentThread().getName());
        } catch (Exception e){
            throw new RuntimeException(e);
        }
        return in;
    }

    public static Mono<String> blockingMethodSingleThread(final String in) {
        return Mono.fromSupplier(() -> blockingWrite(in));
    }
}

BlockingRemoteCall.java

import reactor.core.publisher.Mono;

import java.util.Random;

public class BlockingRemoteCall {

    private final static Random r = new Random();

    static private String blockingWebService(final String in) {
        try {
            // fakes blocking for up to a second
            Thread.sleep((long) (1000 * r.nextFloat()));
            System.out.println("webserver returned: "+in+" on "+Thread.currentThread().getName());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return in;
    }

    public static Mono<String> blockingMethodParallelThread(final String in) {
        return Mono.fromSupplier(() -> blockingWebService(in));
    }
}

TestFluxes.java

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.ParallelFlux;
import reactor.core.scheduler.Schedulers;

public class TestFluxes {

    @Test
    public  void testFluxParallelProcess() throws Exception {

        // Solution by Michael Berry https://stackoverflow.com/a/58709188/329496 Thanks!

        ParallelFlux<String> flux = Flux.range(1, 10).map(i -> i.toString()).parallel().runOn(Schedulers.elastic());

        ParallelFlux<String> pipeline = flux.flatMap(s -> {
            Mono<String> async = BlockingRemoteCall.blockingMethodParallelThread(s);
            String r1 = async.block();
            return BlockingJournal.blockingMethodSingleThread(r1);
        });

        pipeline.sequential().doOnNext(System.out::println).blockLast();

    }
}