Added SParallelFlux.reduce

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

Author: sinwe

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

@@ -57,6 +57,22 @@ class SParallelFlux[T] private(private val jParallelFlux: JParallelFlux[T]) exte
     */
   final def reduce(reducer: (T, T) => T) = Mono(jParallelFlux.reduce(reducer))
 
+  /**
+    * Reduces all values within a 'rail' to a single value (with a possibly different
+    * type) via a reducer function that is initialized on each rail from an
+    * initialSupplier value.
+    * <p>
+    * Note that the same mapper function may be called from multiple threads
+    * concurrently.
+    *
+    * @tparam R the reduced output type
+    * @param initialSupplier the supplier for the initial value
+    * @param reducer         the function to reduce a previous output of reduce (or the initial
+    *                        value supplied) with a current source value.
+    * @return the new [[SParallelFlux]] instance
+    */
+  final def reduce[R](initialSupplier: () => R, reducer: (R, T) => R) = SParallelFlux(jParallelFlux.reduce[R](initialSupplier, reducer))
+
   /**
     * Specifies where each 'rail' will observe its incoming values with possibly
     * work-stealing and a given prefetch amount.

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

@@ -9,7 +9,7 @@ import reactor.test.StepVerifier
 class SParallelFluxTest extends FreeSpec with Matchers {
   "SParallelFlux" - {
     val data = Seq(1, 2, 3)
-    val flux = Flux.just(data.head,data.tail:_*)
+    val flux = Flux.just(data.head, data.tail: _*)
     val fluxParallel: SParallelFlux[Int] = flux.parallel()
     ".asJava should convert as Java ParallelFlux" in {
       fluxParallel.asJava shouldBe a[JParallelFlux[Int]]
@@ -20,7 +20,7 @@ class SParallelFluxTest extends FreeSpec with Matchers {
     }
 
     ".filter should filter elements" in {
-      StepVerifier.create(fluxParallel.filter((i:Int) => i % 2 == 0))
+      StepVerifier.create(fluxParallel.filter((i: Int) => i % 2 == 0))
         .expectNext(2)
         .verifyComplete()
     }
@@ -34,11 +34,34 @@ class SParallelFluxTest extends FreeSpec with Matchers {
         .verifyComplete()
     }
 
-    ".reduce should aggregate the values" in {
-      val mono = fluxParallel.reduce(_ + _)
-      StepVerifier.create(mono)
-        .expectNext(6)
-        .verifyComplete()
+    ".reduce should aggregate the values" - {
+      "without initial supplier" in {
+        val mono = fluxParallel.reduce(_ + _)
+        StepVerifier.create(mono)
+          .expectNext(6)
+          .verifyComplete()
+      }
+      "with initial value should aggregate the values with initial one" ignore {
+        val parallelFlux = fluxParallel.reduce[String](() => "0", (agg, v) => s"$agg-${v.toString}")
+        StepVerifier.create(parallelFlux)
+          .expectNext("0-1")
+          .expectNext("0-2")
+          .expectNext("0-3")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .expectNext("0")
+          .verifyComplete()
+      }
     }
 
     ".sequential should merge the rails" in {