Added Mono.filterWhen

Added Flux.filterWhen

Author: sinwe

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

@@ -1,6 +1,6 @@
 package reactor.core.scala.publisher
 
-import java.lang.{Iterable => JIterable, Long => JLong}
+import java.lang.{Iterable => JIterable, Long => JLong, Boolean => JBoolean}
 import java.util
 import java.util.concurrent.Callable
 import java.util.function.{BiFunction, Consumer, Function, Supplier}
@@ -1231,6 +1231,52 @@ class Flux[T] private[publisher](private[publisher] val jFlux: JFlux[T]) extends
     */
   final def filter(p: T => Boolean) = Flux(jFlux.filter(p))
 
+  /**
+    * Test each value emitted by this [[Flux]] asynchronously using a generated
+    * [[Publisher[Boolean]]] test. A value is replayed if the first item emitted
+    * by its corresponding test is `true`. It is dropped if its test is either
+    * empty or its first emitted value is `false`.
+    * <p>
+    * Note that only the first value of the test publisher is considered, and unless it
+    * is a [[Mono]], test will be cancelled after receiving that first value. Test
+    * publishers are generated and subscribed to in sequence.
+    *
+    * @param asyncPredicate the function generating a [[Publisher]] of [[Boolean]]
+    *                                                         for each value, to filter the Flux with
+    * @return a filtered [[Flux]]
+    */
+  final def filterWhen(asyncPredicate: T => _ <: Publisher[Boolean] with MapablePublisher[Boolean]): Flux[T] = {
+    val asyncPredicateFunction = new Function[T, Publisher[JBoolean]] {
+      override def apply(t: T): Publisher[JBoolean] = asyncPredicate(t).map(Boolean2boolean(_))
+    }
+    Flux(jFlux.filterWhen(asyncPredicateFunction))
+  }
+
+  /**
+    * Test each value emitted by this [[Flux]] asynchronously using a generated
+    * [[Publisher[Boolean]]] test. A value is replayed if the first item emitted
+    * by its corresponding test is `true`. It is dropped if its test is either
+    * empty or its first emitted value is `false`.
+    * <p>
+    * Note that only the first value of the test publisher is considered, and unless it
+    * is a [[Mono]], test will be cancelled after receiving that first value. Test
+    * publishers are generated and subscribed to in sequence.
+    *
+    * @param asyncPredicate the function generating a [[Publisher]] of [[Boolean]]
+    *                                                         for each value, to filter the Flux with
+    * @param bufferSize the maximum expected number of values to hold pending a result of
+    *                   their respective asynchronous predicates, rounded to the next power of two. This is
+    *                   capped depending on the size of the heap and the JVM limits, so be careful with
+    *                   large values (although eg. { @literal 65536} should still be fine). Also serves as
+    *                                                         the initial request size for the source.
+    * @return a filtered [[Flux]]
+    */
+  final def filterWhen(asyncPredicate: T => _ <: Publisher[Boolean] with MapablePublisher[Boolean], bufferSize: Int): Flux[T] = {
+    val asyncPredicateFunction = new Function[T, Publisher[JBoolean]] {
+      override def apply(t: T): Publisher[JBoolean] = asyncPredicate(t).map(Boolean2boolean(_))
+    }
+    Flux(jFlux.filterWhen(asyncPredicateFunction, bufferSize))
+  }
   /**
     * Emit from the fastest first sequence between this publisher and the given publisher
     *

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

@@ -582,9 +582,27 @@ class Mono[T] private(private val jMono: JMono[T]) extends Publisher[T] with Map
     * @param tester the predicate to evaluate
     * @return a filtered [[Mono]]
     */
-  final def filter(tester: T => Boolean) = new Mono[T](
-    jMono.filter(tester)
-  )
+  final def filter(tester: T => Boolean) = Mono[T](jMono.filter(tester))
+
+  /**
+    * If this [[Mono]] is valued, test the value asynchronously using a generated
+    * [[Publisher[Boolean]]] test. The value from the Mono is replayed if the
+    * first item emitted by the test is `true`. It is dropped if the test is
+    * either empty or its first emitted value is false``.
+    * <p>
+    * Note that only the first value of the test publisher is considered, and unless it
+    * is a [[Mono]], test will be cancelled after receiving that first value.
+    *
+    * @param asyncPredicate the function generating a [[Publisher]] of [[Boolean]]
+    *                                                         to filter the Mono with
+    * @return a filtered [[Mono]]
+    */
+  final def filterWhen(asyncPredicate: T => _ <: Publisher[Boolean] with MapablePublisher[Boolean]): Mono[T] = {
+    val asyncPredicateFunction = new Function[T, Publisher[JBoolean]] {
+      override def apply(t: T): Publisher[JBoolean] = asyncPredicate(t).map(Boolean2boolean(_))
+    }
+    Mono(jMono.filterWhen(asyncPredicateFunction))
+  }
 
   /**
     * Transform the item emitted by this [[Mono]] into a Publisher, then forward

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

@@ -1172,6 +1172,21 @@ class FluxTest extends FreeSpec with Matchers with TableDrivenPropertyChecks {
         .verifyComplete()
     }
 
+    ".filterWhen" - {
+      "should replay the value of mono if the first item emitted by the test is true" in {
+        val flux = Flux.just(10, 20, 30).filterWhen((i: Int) => Mono.just(i % 2 == 0))
+        StepVerifier.create(flux)
+          .expectNext(10, 20, 30)
+          .verifyComplete()
+      }
+      "with bufferSize should replay the value of mono if the first item emitted by the test is true" in {
+        val flux = Flux.just(10, 20, 30).filterWhen((i: Int) => Mono.just(i % 2 == 0), 1)
+        StepVerifier.create(flux)
+          .expectNext(10, 20, 30)
+          .verifyComplete()
+      }
+    }
+
     ".firstEmittingWith should emit from the fastest first sequence" in {
       val flux = Flux.just(10, 20, 30).firstEmittingWith(Flux.just(1, 2, 3).delayElements(1 second))
       StepVerifier.create(flux)

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

@@ -830,6 +830,13 @@ class MonoTest extends FreeSpec with Matchers with TableDrivenPropertyChecks {
         .verifyComplete()
     }
 
+    ".filterWhen should replay the value of mono if the first item emitted by the test is true" in {
+      val mono = Mono.just(10).filterWhen((i: Int) => Mono.just(i % 2 == 0))
+      StepVerifier.create(mono)
+      .expectNext(10)
+      .verifyComplete()
+    }
+
     ".flatMap" - {
       "with a single mapper should flatmap the value mapped by the provided mapper" in {
         val flux = Mono.just(1).flatMap(i => Flux.just(i, i * 2))