adds splitBy extension method on scala.collection.Iterator

Iterator#groupUntilChanged constructs an iterator where consecutive
elements of the original iterator are accumulated as long as the output
of a key function for each element doesn't change.

This operation makes sense as soon as you are trying to process an
iterator where you know the elements will be sorted in a certain way and
you need to group them without loading all the data in memory.

For instance
 * processing a file where the ordering is guaranteed but the file
 doesn't fit in the heap,
 * processing a streaming resultset where the underlying database
 guarantees the ordering because of a sort clause.

Author: jeantil

src/main/scala/scala/collection/decorators/IteratorDecorator.scala

@@ -72,4 +72,63 @@ class IteratorDecorator[A](val `this`: Iterator[A]) extends AnyVal {
     loop(immutable.List.empty)
   }
 
+  /**
+    * Constructs an iterator where consecutive elements are accumulated as
+    * long as the output of f for each element doesn't change.
+    * <pre>
+    * Vector(1,2,2,3,3,3,2,2)
+    * .iterator
+    * .splitBy(identity)
+    * .toList
+    * </pre>
+    * produces
+    * <pre>
+    * List(Seq(1),
+    * Seq(2,2),
+    * Seq(3,3,3),
+    * Seq(2,2))
+    * </pre>
+    *
+    * @param f the function to compute a key for an element
+    * @tparam K the type of the computed key
+    * @return an iterator of sequences of the consecutive elements with the
+    *         same key in the original iterator
+    */
+  def splitBy[K](f: A => K): Iterator[Iterable[A]] =
+    new AbstractIterator[Seq[A]] {
+      private var hd: A = _
+      private var hdDefined: Boolean = false
+
+      override def hasNext: Boolean = hdDefined || `this`.hasNext
+
+      override def next(): Seq[A] = {
+        if (hasNext) {
+          val seq = Vector.newBuilder[A]
+          if (hdDefined) {
+            seq += hd
+          } else {
+            hd = `this`.next()
+            hdDefined = true
+            seq += hd
+          }
+          var hadSameKey = true
+          while (`this`.hasNext && hadSameKey) {
+            val el = `this`.next()
+            hdDefined = true
+            if (f(el) == f(hd)) {
+              seq += el
+            } else {
+              hadSameKey = false
+            }
+            hd = el
+          }
+          if (hadSameKey) {
+            hdDefined = false
+          }
+          seq.result()
+        } else {
+          Iterator.empty.next()
+        }
+      }
+    }
 }

src/test/scala/scala/collection/decorators/IteratorDecoratorTest.scala

@@ -0,0 +1,41 @@
+package scala.collection
+package decorators
+
+import org.junit.{Assert, Test}
+
+import scala.util.Try
+
+class IteratorDecoratorTest {
+  @Test
+  def splitByShouldHonorEmptyIterator(): Unit = {
+    val groupedIterator = Iterator.empty.splitBy(identity)
+    Assert.assertFalse(groupedIterator.hasNext)
+    Assert.assertEquals(Try(groupedIterator.next).toString, Try(Iterator.empty.next()).toString)
+  }
+
+  @Test
+  def splitByShouldReturnIteratorOfSingleSeqWhenAllElHaveTheSameKey(): Unit = {
+    val value = Vector("1", "1", "1")
+    val groupedIterator = value.iterator.splitBy(identity)
+    Assert.assertTrue(groupedIterator.hasNext)
+    Assert.assertEquals(groupedIterator.next.toVector, value)
+    Assert.assertFalse(groupedIterator.hasNext)
+    Assert.assertEquals(Try(groupedIterator.next).toString, Try(Iterator.empty.next()).toString)
+  }
+
+  @Test
+  def splitByShouldReturnIteratorOfSeqOfConsecutiveElementsWithTheSameKey(): Unit = {
+    val value = Vector("1", "2", "2", "3", "3", "3", "2", "2")
+    val groupedIterator = value.iterator.splitBy(identity)
+    Assert.assertTrue(groupedIterator.hasNext)
+    Assert.assertEquals(groupedIterator.next.toVector, Vector("1"))
+    Assert.assertTrue(groupedIterator.hasNext)
+    Assert.assertEquals(groupedIterator.next.toVector, Vector("2", "2"))
+    Assert.assertTrue(groupedIterator.hasNext)
+    Assert.assertEquals(groupedIterator.next.toVector, Vector("3", "3", "3"))
+    Assert.assertTrue(groupedIterator.hasNext)
+    Assert.assertEquals(groupedIterator.next.toVector, Vector("2", "2"))
+    Assert.assertFalse(groupedIterator.hasNext)
+    Assert.assertEquals(Try(groupedIterator.next).toString, Try(Iterator.empty.next()).toString)
+  }
+}