Add flatten

Author: danslapman

build.sbt

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

readme.md

@@ -23,7 +23,7 @@ asyncstreams is available via jitpack:
 ```
     resolvers += "jitpack" at "https://jitpack.io"
 
-    libraryDependencies += "com.github.danslapman" %% "asyncstreams" % "0.1"
+    libraryDependencies += "com.github.danslapman" %% "asyncstreams" % "0.3"
 ```
 
 asyncstreams is based on [scala-async](https://github.com/iboltaev/scala-async) ideas.

src/main/scala/asyncstreams/AsyncStream.scala

@@ -1,6 +1,7 @@
 package asyncstreams
 
 import scala.annotation.unchecked.{uncheckedVariance => uV}
+import scala.collection.GenIterable
 import scala.collection.generic.CanBuildFrom
 import scala.concurrent.{ExecutionContext, Future}
 import scala.language.higherKinds
@@ -42,6 +43,16 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[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)
+
+  def flatten[B](implicit asIterable: A => GenIterable[B], executor: ExecutionContext): AsyncStream[B] = {
+    val streamChunk = (p: Pair[A, AsyncStream[A]]) =>
+      concat(generate(asIterable(p.first))(it => if (it.nonEmpty) Future(it.head, it.tail) else ENDF), p.second.flatten)
+
+    AsyncStream(data.flatMap {
+      case END => ENDF
+      case pair => streamChunk(pair).data
+    })
+  }
 }
 
 

src/test/scala/asyncstreams/AsyncStreamTests.scala

@@ -12,7 +12,7 @@ import scalaz.syntax.std.boolean._
 
 
 class AsyncStreamTests extends BaseSuite {
-  private def makeStream(l: List[Int]) = generate(l)(l => ((l.nonEmpty)?(l.head, l.tail)|END).point[Future])
+  private def makeStream[T](l: Iterable[T]) = generate(l)(l => (l.nonEmpty ?(l.head, l.tail)|END).point[Future])
 
   private def makeInfStream = generate(0)(v => Future((v, v + 1)))
 
@@ -73,4 +73,10 @@ class AsyncStreamTests extends BaseSuite {
     Await.ready(task, 10.seconds)
     buffer.to[List] shouldBe 0 :: 1 :: 2 :: 3 :: 4 :: 5 :: 6 :: 7 :: 8 :: 9 :: Nil
   }
+
+  test("flatten") {
+    val stream = makeStream(Vector.range(0, 1000000).grouped(10).to[Vector])
+    val flatStream = stream.flatten
+    wait(flatStream.to[Vector]) shouldBe Vector.range(0, 1000000)
+  }
 }