#12: Added Filter[T]

Author: sinwe

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

@@ -0,0 +1,18 @@
+package reactor.core.scala.publisher
+
+import org.reactivestreams.Publisher
+
+trait Filter[T] { self: Publisher[T] =>
+  /**
+    * Evaluate each accepted value against the given predicate T => Boolean. If the predicate test succeeds, the value is
+    * passed into the new [[Publisher]]. If the predicate test fails, the value is ignored and a request of 1 is
+    * emitted.
+    *
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.5.RELEASE/src/docs/marble/filter.png" alt="">
+    *
+    * @param p the [[Function1]] predicate to test values against
+    * @return a new [[Publisher]] containing only values that pass the predicate test
+    */
+  def filter(f: T => Boolean): Publisher[T]
+}

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

@@ -43,7 +43,7 @@ import scala.concurrent.duration.Duration
   * @see [[Mono]]
   */
 class Flux[T] private[publisher](private[publisher] val jFlux: JFlux[T])
-  extends Publisher[T] with MapablePublisher[T] with OnErrorReturn[T] with FluxLike[T] {
+  extends Publisher[T] with MapablePublisher[T] with OnErrorReturn[T] with FluxLike[T] with Filter [T] {
   override def subscribe(s: Subscriber[_ >: T]): Unit = jFlux.subscribe(s)
 
   /**

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

@@ -63,7 +63,7 @@ import scala.util.{Failure, Success, Try}
   * @see Flux
   */
 class Mono[T] private(private val jMono: JMono[T])
-  extends Publisher[T] with MapablePublisher[T] with OnErrorReturn[T] with MonoLike[T] {
+  extends Publisher[T] with MapablePublisher[T] with OnErrorReturn[T] with MonoLike[T] with Filter[T] {
   override def subscribe(s: Subscriber[_ >: T]): Unit = jMono.subscribe(s)
 
   /**