Merge branch 'master' into sreactor

Author: sinwe

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

@@ -2392,9 +2392,9 @@ class Flux[T] private[publisher](private[publisher] val jFlux: JFlux[T])
     * <p>
     * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.5.RELEASE/src/docs/marble/parallel.png" alt="">
     *
-    * @return a new [[ParallelFlux]] instance
+    * @return a new [[SParallelFlux]] instance
     */
-  final def parallel() = ParallelFlux(jFlux.parallel())
+  final def parallel() = SParallelFlux(jFlux.parallel())
 
   /**
     * Prepare to consume this [[Flux]] on parallelism number of 'rails'
@@ -2404,9 +2404,9 @@ class Flux[T] private[publisher](private[publisher] val jFlux: JFlux[T])
     * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.5.RELEASE/src/docs/marble/parallel.png" alt="">
     *
     * @param parallelism the number of parallel rails
-    * @return a new [[ParallelFlux]] instance
+    * @return a new [[SParallelFlux]] instance
     */
-  final def parallel(parallelism: Int): ParallelFlux[T] = ParallelFlux(jFlux.parallel(parallelism))
+  final def parallel(parallelism: Int): SParallelFlux[T] = SParallelFlux(jFlux.parallel(parallelism))
 
   /**
     * Prepare to consume this [[Flux]] on parallelism number of 'rails'
@@ -2418,9 +2418,9 @@ class Flux[T] private[publisher](private[publisher] val jFlux: JFlux[T])
     *
     * @param parallelism the number of parallel rails
     * @param prefetch    the number of values to prefetch from the source
-    * @return a new [[ParallelFlux]] instance
+    * @return a new [[SParallelFlux]] instance
     */
-  final def parallel(parallelism: Int, prefetch: Int) = ParallelFlux(jFlux.parallel(parallelism, prefetch))
+  final def parallel(parallelism: Int, prefetch: Int) = SParallelFlux(jFlux.parallel(parallelism, prefetch))
 
   /**
     * Prepare a [[ConnectableFlux]] which shares this [[Flux]] sequence and dispatches values to

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

@@ -0,0 +1,54 @@
+package reactor.core.scala.publisher
+
+import java.util
+import java.util.function.Supplier
+
+import org.reactivestreams.Publisher
+import reactor.core.publisher.{ParallelFlux => JParallelFlux}
+import reactor.util.concurrent.Queues
+
+class SParallelFlux[T] private(private val jParallelFlux: JParallelFlux[T]) {
+
+  /**
+    * Perform a fluent transformation to a value via a converter function which receives
+    * this ParallelFlux.
+    *
+    * @tparam U the output value type
+    * @param converter the converter function from [[SParallelFlux]] to some type
+    * @return the value returned by the converter function
+    */
+  final def as[U](converter: SParallelFlux[T] => U): U = jParallelFlux.as((t: JParallelFlux[T]) => converter(SParallelFlux(t)))
+
+  def asJava: JParallelFlux[T] = jParallelFlux
+}
+
+object SParallelFlux {
+  def apply[T](jParallelFlux: JParallelFlux[T]) = new SParallelFlux[T](jParallelFlux)
+
+  /**
+    * Take a Publisher and prepare to consume it on multiple 'rails' (one per CPU core)
+    * in a round-robin fashion.
+    *
+    * @tparam T the value type
+    * @param source        the source Publisher
+    * @param parallelism   the number of parallel rails
+    * @param prefetch      the number of values to prefetch from the source
+    * @param queueSupplier the queue structure supplier to hold the prefetched values
+    *                      from the source until there is a rail ready to process it.
+    * @return the [[SParallelFlux]] instance
+    */
+  def from[T](source: Publisher[_ <: T],
+              parallelism: Int = Runtime.getRuntime.availableProcessors(),
+              prefetch: Int = Queues.SMALL_BUFFER_SIZE,
+              queueSupplier: Supplier[util.Queue[T]] = Queues.small()) = SParallelFlux(JParallelFlux.from(source, parallelism, prefetch, queueSupplier))
+
+  /**
+    * Wraps multiple Publishers into a [[SParallelFlux]] which runs them in parallel and
+    * unordered.
+    *
+    * @tparam T the value type
+    * @param publishers the array of publishers
+    * @return the [[SParallelFlux]] instance
+    */
+  def fromPublishers[T](publishers: Publisher[T]*) = SParallelFlux(JParallelFlux.from(publishers: _*))
+}

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

@@ -0,0 +1,16 @@
+package reactor.core.scala.publisher
+
+import org.scalatest.{FreeSpec, Matchers}
+import reactor.core.publisher.{ParallelFlux => JParallelFlux, Flux => JFlux}
+
+class SParallelFluxTest extends FreeSpec with Matchers {
+  "SParallelFlux" - {
+    ".asJava should convert as Java ParallelFlux" in {
+      Flux.just(1, 2, 3).parallel().asJava shouldBe a[JParallelFlux[Int]]
+    }
+
+    ".apply should convert Java ParallelFlux into SParallelFlux" in {
+      SParallelFlux(JFlux.just(1, 2, 3).parallel()).asJava shouldBe a[JParallelFlux[Int]]
+    }
+  }
+}