Added SParallelFlux.sequential

Signed-off-by: Winarto Zhao <[email protected]>

Author: sinwe

src/main/scala/reactor/core/scala/publisher/SParallelFlux.scala

@@ -66,6 +66,17 @@ class SParallelFlux[T] private(private val jParallelFlux: JParallelFlux[T]) exte
     */
   final def runOn(scheduler: Scheduler, prefetch: Int = Queues.SMALL_BUFFER_SIZE) = SParallelFlux(jParallelFlux.runOn(scheduler, prefetch))
 
+
+  /**
+    * Merges the values from each 'rail' in a round-robin or same-order fashion and
+    * exposes it as a regular Publisher sequence, running with a give prefetch value for
+    * the rails.
+    *
+    * @param prefetch the prefetch amount to use for each rail
+    * @return the new Flux instance
+    */
+  final def sequential(prefetch: Int = Queues.SMALL_BUFFER_SIZE) = Flux(jParallelFlux.sequential(prefetch))
+
   /**
     * Subscribes to this [[SParallelFlux]] by providing an onNext, onError,
     * onComplete and onSubscribe callback and triggers the execution chain for all

src/test/scala/reactor/core/scala/publisher/SParallelFluxTest.scala

@@ -34,6 +34,15 @@ class SParallelFluxTest extends FreeSpec with Matchers {
         .verifyComplete()
     }
 
+    ".sequential should merge the rails" in {
+      val expected = data.map(_.toString)
+      StepVerifier.create(fluxParallel.map(i => i.toString).sequential())
+        .expectNextMatches((i: String) => expected.contains(i))
+        .expectNextMatches((i: String) => expected.contains(i))
+        .expectNextMatches((i: String) => expected.contains(i))
+        .verifyComplete()
+    }
+
     ".runOn should run on different thread" in {
       val scheduler = spy(Schedulers.parallel())
       StepVerifier.create(flux.parallel(2).runOn(scheduler))