Java class full of static methods that give streams from collections.

Author: Ichoran

README.md

@@ -117,6 +117,49 @@ class Test {
 ```
 
 
-## Future work
-  - Converters for `java.util.stream`
-  - [`Spliterator`](https://docs.oracle.com/javase/8/docs/api/java/util/Spliterator.html)s for Scala collections
+## Converters from Scala collections to Java 8 Streams
+
+Scala collections gain `seqStream` and `parStream` as extension methods that produce a Java 8 Stream
+running sequentially or in parallel, respectively.  These are automatically specialized to a primitive
+type if possible.  For instance, `List(1,2).seqStream` produces an `IntStream`.  Maps additionally have
+`seqKeyStream`, `seqValueStream`, `parKeyStream`, and `parValueStream` methods.
+
+Scala collections also gain `accumulate` and `stepper` methods that produce utility collections that
+can be useful when working with Java 8 Streams.  `accumulate` produces an `Accumulator` or its primitive
+counterpart (`DoubleAccumulator`, etc.), which is a low-level collection designed for efficient collection
+and dispatching of results to and from Streams.  Unlike most collections, it can contain more than
+`Int.MaxValue` elements.  `stepper` produces a `Stepper` which is a fusion of `Spliterator` and `Iterator`.
+`Stepper`s underlie the Scala collections' instances of Java 8 Streams.
+
+Java 8 Streams gain `toScala[Coll]` and `accumulate` methods, to make it easy to produce Scala collections
+or Accumulators, respectively, from Java 8 Streams. For instance, `myStream.to[Vector]` will collect the
+contents of a Stream into a `scala.collection.immutable.Vector`.  Note that standard sequential builders
+are used for collections, so this is best done to gather the results of an expensive computation.
+
+Finally, there is a Java class, `ScalaStreamer`, that has a series of `from` methods that can be used to
+obtain Java 8 Streams from Scala collections from within Java.
+
+#### Scala Usage Example
+
+```scala
+import scala.compat.java8.StreamConverters._
+
+object Test {
+  val m = collection.immutable.HashMap("fish" -> 2, "bird" -> 4)
+  val s = m.parValueStream.sum          // 6, potientially computed in parallel
+  val t = m.seqKeyStream.toScala[List]  // List("fish", "bird")
+  val a = t.accumulate                  // Accumulator[(String, Int)]
+
+  val n = a.splitter.fold(0)(_ + _._1.length) +
+          a.parStream.count             // 8 + 2 = 10
+
+  val b = java.util.Arrays.stream(Array(2L, 3L, 4L)).
+          accumulate                    // LongAccumulator
+}
+```
+
+#### Java Usage Example
+
+```java
+// TODO -- write converter and create example
+```

src/main/java/scala/compat/java8/ScalaStreaming.java

@@ -137,6 +137,10 @@ trait Priority1StreamConverters extends Priority2StreamConverters {
   * 
   * Scala collections gain extension methods `seqStream` and
   * `parStream` that allow them to be used as the source of a `Stream`.
+  * Some collections either intrinsically cannot be paralellized, or
+  * could be but an efficient implementation is missing.  It this case,
+  * only `seqStream` is provided.  If a collection cannot be stepped over
+  * at all (e.g. `Traversable`), then it gains neither method.
   *
   * `Array` also gains `seqStream` and `parStream` methods, and calling those
   * on `Array[Double]`, `Array[Int]`, or `Array[Long]` will produce the
@@ -152,10 +156,32 @@ trait Priority1StreamConverters extends Priority2StreamConverters {
   * have custom accumulators with improved performance.
   *
   * Accumulators have `toArray`, `toList`, `iterator`, and `to[_]` methods
-  * to convert to standard Scala collections.
+  * to convert to standard Scala collections.  Note that if you wish to
+  * create an array from a `Stream`, going through an `Accumulator` is
+  * not the most efficient option: just create the `Array` directly.
   *
-  * Example:
-  * ```
+  * Internally, Scala collections implement a hybrid of `Iterator` and
+  * `java.util.Spliterator` to implement `Stream` compatibility; these
+  * are called `Stepper`s.  In particular, they can test for the presence
+  * of a next element using `hasStep`, can retrieve the next value with
+  * `nextStep`, or can optionally retrieve and operate on a value if present
+  * with `tryStep`, which works like `tryAdvance` in `java.util.Spliterator`.
+  *
+  * Every Scala collection that can be stepped
+  * through has a `stepper` method implicitly provided.  In addition,
+  * maps have `keyStepper` and `valueStepper` methods.  A limited number
+  * of collections operations are defined on `Stepper`s, including conversion
+  * to Scala collections with `to` or accumulation via `accumulate`.
+  * `Stepper`s also implement `seqStream` and `parStream` to generate `Stream`s.
+  * These are provided regardless of whether a `Stepper` can efficiently
+  * subdivide itself for parallel processing (though one can check for the
+  * presence of the `EfficientSubstep` trait to know that parallel execution will
+  * not be limited by long sequential searching steps, and one can call
+  * `anticipateParallelism` to warn a `Stepper` that it will be used in a parallel
+  * context and thus may wish to make different tradeoffs).
+  *
+  * Examples:
+  * {{{
   * import scala.compat.java8.StreamConverers._
   *
   * val s = Vector(1,2,3,4).parStream    // Stream[Int]
@@ -165,7 +191,9 @@ trait Priority1StreamConverters extends Priority2StreamConverters {
   *
   * val t = Array(2.0, 3.0, 4.0).parStream               // DoubleStream
   * val q = t.toScala[scala.collection.immutable.Queue]  // Queue[Double]
-  * ```
+  *
+  * val x = List(1L, 2L, 3L, 4L).stepper.parStream.sum   // 10, potentially computed in parallel
+  * }}}
   */
 object StreamConverters
 extends Priority1StreamConverters
@@ -273,4 +301,28 @@ with converterImpls.Priority1AccumulatorConverters
       }
     }
   }
+
+  implicit val accumulateDoubleStepper = new AccumulatesFromStepper[Double, DoubleAccumulator] {
+    def apply(stepper: Stepper[Double]) = {
+      val a = new DoubleAccumulator
+      while (stepper.hasStep) a += stepper.nextStep
+      a
+    }
+  }
+
+  implicit val accumulateIntStepper = new AccumulatesFromStepper[Int, IntAccumulator] {
+    def apply(stepper: Stepper[Int]) = {
+      val a = new IntAccumulator
+      while (stepper.hasStep) a += stepper.nextStep
+      a
+    }
+  }
+
+  implicit val accumulateLongStepper = new AccumulatesFromStepper[Long, LongAccumulator] {
+    def apply(stepper: Stepper[Long]) = {
+      val a = new LongAccumulator
+      while (stepper.hasStep) a += stepper.nextStep
+      a
+    }
+  }
 }

src/main/scala/scala/compat/java8/StreamConverters.scala

@@ -5,6 +5,10 @@ import language.implicitConversions
 import scala.compat.java8.collectionImpl._
 import scala.compat.java8.runtime._
 
+trait AccumulatesFromStepper[@specialized(Double, Int, Long) A, Acc <: AccumulatorLike[A, Acc]] {
+  def apply(stepper: Stepper[A]): Acc
+}
+
 final class CollectionCanAccumulate[A](private val underlying: TraversableOnce[A]) extends AnyVal {
   def accumulate: Accumulator[A] = {
     val a = new Accumulator[A]

src/main/scala/scala/compat/java8/collectionImpl/Accumulates.scala

@@ -214,6 +214,13 @@ object Accumulator {
 
   /** A `BiConsumer` that merges `Accumulator`s, suitable for use with `java.util.stream.Stream`'s `collect` method. */
   def merger[A] = new java.util.function.BiConsumer[Accumulator[A], Accumulator[A]]{ def accept(a1: Accumulator[A], a2: Accumulator[A]) { a1 drain a2 } }
+
+  /** Builds an `Accumulator` from any `TraversableOnce` */
+  def from[A](source: TraversableOnce[A]) = {
+    val a = new Accumulator[A]
+    source.foreach(a += _)
+    a
+  }
 }
 
 private[java8] class AccumulatorStepper[A](private val acc: Accumulator[A]) extends AnyStepper[A] {

src/main/scala/scala/compat/java8/collectionImpl/Accumulator.scala

@@ -2,6 +2,8 @@ package scala.compat.java8.converterImpls
 
 import language.implicitConversions
 
+import scala.compat.java8.collectionImpl._
+
 trait Priority3AccumulatorConverters {
   implicit def collectionCanAccumulate[A](underlying: TraversableOnce[A]) = new CollectionCanAccumulate[A](underlying)
 }
@@ -17,4 +19,17 @@ trait Priority1AccumulatorConverters extends Priority2AccumulatorConverters {
   implicit def accumulateDoubleArray(underlying: Array[Double]) = new AccumulateDoubleArray(underlying)
   implicit def accumulateIntArray(underlying: Array[Int]) = new AccumulateIntArray(underlying)
   implicit def accumulateLongArray(underlying: Array[Long]) = new AccumulateLongArray(underlying)
+
+  implicit def accumulateAnyStepper[A]: AccumulatesFromStepper[A, Accumulator[A]] =
+    PrivateAccumulatorConverters.genericAccumulateAnyStepper.asInstanceOf[AccumulatesFromStepper[A, Accumulator[A]]]
+}
+
+private[java8] object PrivateAccumulatorConverters {
+  val genericAccumulateAnyStepper: AccumulatesFromStepper[Any, Accumulator[Any]] = new AccumulatesFromStepper[Any, Accumulator[Any]] {
+    def apply(stepper: Stepper[Any]) = {
+      val a = new Accumulator[Any]
+      while (stepper.hasStep) a += stepper.nextStep
+      a
+    }
+  }
 }

src/main/scala/scala/compat/java8/collectionImpl/AccumulatorConverters.scala

@@ -211,6 +211,13 @@ object DoubleAccumulator {
 
   /** A `BiConsumer` that merges `DoubleAccumulator`s, suitable for use with `java.util.stream.DoubleStream`'s `collect` method.  Suitable for `Stream[Double]` also. */
   def merger = new java.util.function.BiConsumer[DoubleAccumulator, DoubleAccumulator]{ def accept(a1: DoubleAccumulator, a2: DoubleAccumulator) { a1 drain a2 } }
+
+  /** Builds a `DoubleAccumulator` from any `Double`-valued `TraversableOnce` */
+  def from[A](source: TraversableOnce[Double]) = {
+    val a = new DoubleAccumulator
+    source.foreach(a += _)
+    a
+  }
 }
 
 private[java8] class DoubleAccumulatorStepper(private val acc: DoubleAccumulator) extends DoubleStepper {

src/main/scala/scala/compat/java8/collectionImpl/DoubleAccumulator.scala

@@ -218,6 +218,13 @@ object IntAccumulator {
 
   /** A `BiConsumer` that merges `IntAccumulator`s, suitable for use with `java.util.stream.IntStream`'s `collect` method.  Suitable for `Stream[Int]` also. */
   def merger = new java.util.function.BiConsumer[IntAccumulator, IntAccumulator]{ def accept(a1: IntAccumulator, a2: IntAccumulator) { a1 drain a2 } }
+
+  /** Builds an `IntAccumulator` from any `Int`-valued `TraversableOnce` */
+  def from[A](source: TraversableOnce[Int]) = {
+    val a = new IntAccumulator
+    source.foreach(a += _)
+    a
+  }
 }
 
 private[java8] class IntAccumulatorStepper(private val acc: IntAccumulator) extends IntStepper {

src/main/scala/scala/compat/java8/collectionImpl/IntAccumulator.scala

@@ -212,6 +212,13 @@ object LongAccumulator {
 
   /** A `BiConsumer` that merges `LongAccumulator`s, suitable for use with `java.util.stream.LongStream`'s `collect` method.  Suitable for `Stream[Long]` also. */
   def merger = new java.util.function.BiConsumer[LongAccumulator, LongAccumulator]{ def accept(a1: LongAccumulator, a2: LongAccumulator) { a1 drain a2 } }
+
+  /** Builds a `LongAccumulator` from any `Long`-valued `TraversableOnce` */
+  def from[A](source: TraversableOnce[Long]) = {
+    val a = new LongAccumulator
+    source.foreach(a += _)
+    a
+  }
 }
 
 private[java8] class LongAccumulatorStepper(private val acc: LongAccumulator) extends LongStepper {

src/main/scala/scala/compat/java8/collectionImpl/LongAccumulator.scala

@@ -36,7 +36,14 @@ import java.util.Spliterator
   * println(s.hasStep)                      //  Prints `false`
   * }}}
   */
-trait Stepper[@specialized(Double, Int, Long) A] extends StepperLike[A, Stepper[A]] {}
+trait Stepper[@specialized(Double, Int, Long) A] extends StepperLike[A, Stepper[A]] {
+  /** Drains the contents of this stepper into an `Accumulator` or specialized variant thereof as appropriate.
+    * This is a terminal operation.
+    *
+    * Note: accumulation will occur sequentially.  To accumulate in parallel, use a `Stream` (i.e. `.parStream.accumulate`).
+    */
+  def accumulate[Acc <: AccumulatorLike[A, Acc]](implicit accer: scala.compat.java8.converterImpls.AccumulatesFromStepper[A, Acc]) = accer(this)  
+}
 
 /** An (optional) marker trait that indicates that a `Stepper` can call `substep` with
   * at worst O(log N) time and space complexity, and that the division is likely to
@@ -159,6 +166,13 @@ trait StepperLike[@specialized(Double, Int, Long) A, +CC] { self =>
     def hasNext = self.hasStep
     def next = self.nextStep
   }
+
+  /** Returns a Scala collection of the type requested. */
+  def to[Coll[_]](implicit cbf: collection.generic.CanBuildFrom[Nothing, A, Coll[A]]): Coll[A] = {
+    val b = cbf()
+    while (hasStep) b += nextStep
+    b.result()
+  }
 }