Add foreach and foreachF operations

Author: danslapman

build.sbt

@@ -1,6 +1,6 @@
 name := "asyncstreams"
 
-version := "0.1"
+version := "0.2"
 
 scalaVersion := "2.11.8"
 

src/main/scala/asyncstreams/AsyncStream.scala

@@ -36,6 +36,12 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
       case END => END
       case p => Pair(p.first, p.second.take(n - 1))
     })
+
+  def foreach[U](f: (A) => U)(implicit executor: ExecutionContext): Future[Unit] =
+    foldLeft(())((_: Unit, a: A) => {f(a); ()})
+
+  def foreachF[U](f: (A) => Future[U])(implicit executor: ExecutionContext): Future[Unit] =
+    foldLeft(Future(()))((fu: Future[Unit], a: A) => fu.flatMap(_ => f(a)).map(_ => ())).flatMap(u => u)
 }
 
 

src/test/scala/asyncstreams/AsyncStreamTests.scala

@@ -5,6 +5,7 @@ import scala.concurrent.duration._
 import scala.concurrent.ExecutionContext.Implicits.global
 import asyncstreams.AsyncStream._
 
+import scala.collection.mutable.ArrayBuffer
 import scalaz.std.scalaFuture._
 import scalaz.syntax.monad._
 import scalaz.syntax.std.boolean._
@@ -56,4 +57,20 @@ class AsyncStreamTests extends BaseSuite {
     val r = makeInfStream.takeWhile(_ < 1000000)
     wait(r.to[List]) shouldBe (0 to 999999)
   }
+
+  test("foreach") {
+    val stream = makeInfStream.take(10)
+    val buffer = ArrayBuffer[Int]()
+    val task = stream.foreach(i => buffer += i)
+    Await.ready(task, 10.seconds)
+    buffer.to[List] shouldBe 0 :: 1 :: 2 :: 3 :: 4 :: 5 :: 6 :: 7 :: 8 :: 9 :: Nil
+  }
+
+  test("foreachF") {
+    val stream = makeInfStream.take(10)
+    val buffer = ArrayBuffer[Int]()
+    val task = stream.foreachF(i => Future(buffer += i))
+    Await.ready(task, 10.seconds)
+    buffer.to[List] shouldBe 0 :: 1 :: 2 :: 3 :: 4 :: 5 :: 6 :: 7 :: 8 :: 9 :: Nil
+  }
 }