title: out of order processing of a Flux of input using Reactor
body: <p>I am trying to figure out how to take a Flux that is a sequence of input data, in <em>parallel</em> run them through a blocking call that may reorder the sequence, then run the reordered data through a second single-threaded blocking call. The idea being that the final single-threaded call is recording the reordered parallel work output onto disk. The ultimate aim of what I am trying to do is that the parallel algorithm is a consensus algorithm that will determine the actual ordering of the data input. The single-threaded write is simulating processing things in the order determined by the consensus algorithm. </p>
  
  <p>Looking at <a href="https://itnext.io/how-to-make-legacy-code-reactive-2debcb3d0a40" rel="nofollow noreferrer">this article</a> it suggests I should convert my blocking calls into Mono that run on a scheduler that either gives me parallel or single-threaded processing:</p>
  
  <pre class="lang-java prettyprint-override"><code>public class BlockingRemoteCall {
  
      private final static Random r = new Random();
  
      private final static Scheduler scheduler = Schedulers.newParallel("myWebservice", 10);
  
      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&lt;String&gt; blockingMethodParallelThread(final String in) {
          return Mono.fromSupplier(() -&gt; blockingWebService(in))
                  .subscribeOn(scheduler);
      }
  }
  
  public class BlockingJournal {
  
      private final static Scheduler scheduler = Schedulers.newSingle("myJournal");
  
      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&lt;String&gt; blockingMethodSingleThread(final String in) {
          return Mono.fromSupplier(() -&gt; blockingWrite(in))
                  .subscribeOn(scheduler);
      }
  }
  
  </code></pre>
  
  <p>I have been trying to take a Flux of integers and somehow map or flatMap though these methods but I am unable to get anything to log. Here is my latest attempt: </p>
  
  <pre class="lang-java prettyprint-override"><code>        final Scheduler parallelScheduler = Schedulers.newParallel("p");
          final Scheduler singleScheduler = Schedulers.single();
  
          Flux&lt;String&gt; flux = Flux.range(1, 10).map(i -&gt; i.toString()).publishOn(parallelScheduler);
  
          Flux&lt;String&gt; pipeline = flux.map(s-&gt;{
              Mono&lt;String&gt; async = BlockingRemoteCall.blockingMethodParallelThread(s);
              String r1 = async.block();
              Mono&lt;String&gt; r2 = BlockingJournal.blockingMethodSingleThread(r1);
              return r2.block();
          });
  
          pipeline.subscribeOn(singleScheduler).doOnNext(System.out::println).blockLast();
  </code></pre>
  
  <p>That doesn't actually output anything but whenever I have been able to generate any output I only ever see <code>println</code> statements showing that the stream of data was processed in order on one thread. What I am hoping to see is that the arbitrary delays in the call to <code>blockingMethodParallelThread(s)</code> causing the sequence of input to be logged out-of-order. </p>
  
  <p>How do I set things up so that a Flux of input (eventually bubbling up from reactor-netty input) can be processed in parallel, with reordering, then finally processed sequentially, preserving the reordering? where the reordering is due to making blocking calls in parallel? Thanks!</p>