Added Flux.fromStream

Author: sinwe

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

@@ -5,7 +5,7 @@ import java.util
 import java.util.concurrent.Callable
 import java.util.function.{BiFunction, Consumer, Function, Supplier}
 import java.util.logging.Level
-import java.util.{Comparator, List => JList}
+import java.util.{Comparator, stream, List => JList}
 
 import org.reactivestreams.{Publisher, Subscriber, Subscription}
 import reactor.core.Disposable
@@ -4591,16 +4591,37 @@ object Flux {
 
   /**
     * Create a [[Flux]] that emits the items contained in the provided [[Stream]].
-    * A new iterator will be created for each subscriber.
+    * Keep in mind that a [[Stream]] cannot be re-used, which can be problematic in
+    * case of multiple subscriptions or re-subscription (like with [[Flux.repeat()]] or
+    * [[Flux.retry]]). The [[Stream]] is [[Stream.close closed]] automatically
+    * by the operator on cancellation, error or completion.
     * <p>
-    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.5.RELEASE/src/docs/marble/fromstream.png" alt="">
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.1.RELEASE/src/docs/marble/fromstream.png" alt="">
     * <p>
     *
     * @param s the [[Stream]] to read data from
-    * @tparam T the [[Stream]] type to flux
+    * @tparam T The type of values in the source [[Stream]] and resulting Flux
     * @return a new [[Flux]]
     */
-  def fromStream[T](s: Stream[T]): Flux[T] = Flux.fromIterable(s)
+  def fromStream[T](s: Stream[T]) = Flux(JFlux.fromStream(s))
+
+  /**
+    * Create a [[Flux]] that emits the items contained in a [[Stream]] created by
+    * the provided [[Function0]] for each subscription. The [[Stream]] is
+    * [[Stream.close closed]] automatically by the operator on cancellation, error
+    * or completion.
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.1.RELEASE/src/docs/marble/fromstream.png" alt="">
+    * <p>
+    *
+    * @param streamSupplier the [[Function0]] that generates the [[Stream]] from
+    *                                   which to read data
+    * @tparam T The type of values in the source [[Stream]] and resulting Flux
+    * @return a new [[Flux]]
+    */
+  def fromStream[T](streamSupplier: () => Stream[T]) = Flux(JFlux.fromStream[T](new Supplier[stream.Stream[_ <: T]] {
+    override def get() = streamSupplier()
+  }))
 
   /**
     * Generate signals one-by-one via a consumer callback.

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

@@ -2,17 +2,19 @@ package reactor.core.scala
 
 import java.lang.{Boolean => JBoolean, Iterable => JIterable, Long => JLong}
 import java.time.{Duration => JDuration}
-import java.util.Optional
+import java.util.{Optional, Spliterator, Spliterators}
 import java.util.Optional.empty
 import java.util.concurrent.Callable
 import java.util.function.{BiConsumer, BiFunction, BiPredicate, BooleanSupplier, Consumer, Function, LongConsumer, Predicate, Supplier}
+import java.util.stream.{StreamSupport, Stream => JStream}
 
 import org.reactivestreams.Publisher
 import reactor.core.publisher.{Flux => JFlux, Mono => JMono}
 import reactor.util.function.{Tuple2, Tuple3, Tuple4, Tuple5, Tuple6}
 
 import scala.concurrent.duration.Duration
 import scala.language.implicitConversions
+import scala.collection.JavaConverters._
 
 /**
   * Created by winarto on 12/31/16.
@@ -140,7 +142,6 @@ Uncomment this when used. It is not used for now and reduce the code coverage
   }
 
   implicit def scalaIterable2JavaIterable[T](scalaIterable: Iterable[T]): JIterable[T] = {
-    import scala.collection.JavaConverters._
     scalaIterable.asJava
   }
 
@@ -161,4 +162,8 @@ Uncomment this when used. It is not used for now and reduce the code coverage
   }
 
   implicit def javaOptional2ScalaOption[T](jOptional: Optional[T]): Option[T] = if(jOptional.isPresent) Some(jOptional.get()) else None
+
+  implicit def scalaStream2JavaStream[T](stream: Stream[T]): JStream[T] = {
+    StreamSupport.stream(Spliterators.spliteratorUnknownSize[T](stream.toIterator.asJava, Spliterator.NONNULL), false)
+  }
 }

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

@@ -258,11 +258,19 @@ class FluxTest extends FreeSpec with Matchers with TableDrivenPropertyChecks {
         .verifyComplete()
     }
 
-    ".fromStream should create flux that emit items contained in the provided stream" in {
-      val flux = Flux.fromStream(Stream(1, 2, 3))
-      StepVerifier.create(flux)
-        .expectNext(1, 2, 3)
-        .verifyComplete()
+    ".fromStream" - {
+      "should create flux that emit items contained in the provided stream" in {
+        val flux = Flux.fromStream(Stream(1, 2, 3))
+        StepVerifier.create(flux)
+          .expectNext(1, 2, 3)
+          .verifyComplete()
+      }
+      "with supplier should create flux that emit items contained in the supplier" in {
+        val flux = Flux.fromStream(() => Stream(1, 2, 3))
+        StepVerifier.create(flux)
+          .expectNext(1, 2, 3)
+          .verifyComplete()
+      }
     }
 
     ".generate" - {