Added SParallelFlux.filter

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

Author: sinwe

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

@@ -21,6 +21,17 @@ class SParallelFlux[T] private(private val jParallelFlux: JParallelFlux[T]) exte
     */
   final def as[U](converter: SParallelFlux[T] => U): U = jParallelFlux.as((t: JParallelFlux[T]) => converter(SParallelFlux(t)))
 
+  /**
+    * Filters the source values on each 'rail'.
+    * <p>
+    * Note that the same predicate may be called from multiple threads concurrently.
+    *
+    * @param predicate the function returning true to keep a value or false to drop a
+    *                  value
+    * @return the new [[SParallelFlux]] instance
+    */
+  final def filter(predicate: SPredicate[T]) = SParallelFlux(jParallelFlux.filter(predicate))
+
   /**
     * Maps the source values on each 'rail' to another value.
     * <p>

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

@@ -19,6 +19,12 @@ class SParallelFluxTest extends FreeSpec with Matchers {
       SParallelFlux(JFlux.just(1, 2, 3).parallel()).asJava shouldBe a[JParallelFlux[Int]]
     }
 
+    ".filter should filter elements" in {
+      StepVerifier.create(fluxParallel.filter(i => i % 2 == 0))
+        .expectNext(2)
+        .verifyComplete()
+    }
+
     ".map should map from T to U" in {
       val expected = data.map(_.toString)
       StepVerifier.create(fluxParallel.map(i => i.toString))