adds groupUntilChanged 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,44 @@ 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
+    *     .groupUntilChanged(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 groupUntilChanged[K](f: A => K): Iterator[Iterable[A]] =
+    new AbstractIterator[Seq[A]] {
+      private var hd: Option[A] = `this`.nextOption()
+      override def hasNext: Boolean = hd.isDefined
+
+      override def next(): Seq[A] = {
+        hd match {
+          case None => Iterator.empty.next()
+          case Some(head) =>
+            hd = `this`.nextOption()
+            var seq = mutable.Buffer(head)
+            while (hd.exists(el => f(el) == f(head))) {
+              seq = seq ++ hd
+              hd = `this`.nextOption()
+            }
+            seq.toVector
+        }
+      }
+    }
 }

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

@@ -0,0 +1,37 @@
+package scala.collection
+package decorators
+
+import scala.util.Try
+
+import org.junit.{Assert, Test}
+
+class IteratorDecoratorTest {
+  @Test
+  def groupUntilChangedShouldHonorEmptyIterator(): Unit ={
+    val groupedIterator = Iterator.empty.groupUntilChanged(identity)
+    Assert.assertFalse(groupedIterator.hasNext)
+    Assert.assertEquals(Try(groupedIterator.next).toString, Try(Iterator.empty.next()).toString)  }
+  @Test
+  def groupUntilChangedShouldReturnSingleSeqWhenAllElHaveTheSameKey(): Unit ={
+    val value = Vector("1", "1", "1")
+    val groupedIterator = value.iterator.groupUntilChanged(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 groupUntilChangedShouldReturnSeqOfConsecutiveElementsWithTheSameKey(): Unit ={
+    val value = Vector("1", "2","2","3","3","3","2","2")
+    val groupedIterator = value.iterator.groupUntilChanged(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)  }
+}