Initial implementation

Author: danslapman

.gitignore

@@ -15,3 +15,4 @@ project/plugins/project/
 # Scala-IDE specific
 .scala_dependencies
 .worksheet
+.idea/

build.sbt

@@ -0,0 +1,10 @@
+name := "asyncstreams"
+
+version := "1.0"
+
+scalaVersion := "2.11.8"
+
+libraryDependencies ++= Seq(
+  "org.scalaz" %% "scalaz-core" % "7.2.6",
+  "org.scalatest" %% "scalatest" % "3.0.0" % Test
+)

copying.txt

@@ -0,0 +1,13 @@
+            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
+                    Version 2, December 2004
+
+ Copyright (C) 2016 Daniil Smirnov <danslapman@gmail.com>
+
+ Everyone is permitted to copy and distribute verbatim or modified
+ copies of this license document, and changing it is allowed as long
+ as the name is changed.
+
+            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
+   TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
+
+  0. You just DO WHAT THE FUCK YOU WANT TO.

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 0.13.12

project/plugins.sbt

@@ -0,0 +1 @@
+logLevel := Level.Warn

src/main/scala/asyncstreams/AsyncStream.scala

@@ -0,0 +1,47 @@
+package asyncstreams
+
+import scala.concurrent.{ExecutionContext, Future}
+import scalaz.std.scalaFuture._
+import scalaz.syntax.std.option._
+import scalaz.syntax.monad._
+import scalaz.OptionT.{optionT => opT}
+
+case class AsyncStream[A](data: Future[Chunk[A, AsyncStream[A]]]) {
+  import AsyncStream._
+
+  def foldLeft[B](start: B)(f: (B, A) => B)(implicit executor: ExecutionContext): Future[B] = {
+    def impl(d: Future[Chunk[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
+      d.flatMap(chunk => chunk.map(p => impl(p.second.data, acc map (b => f(b, p.first)))).getOrElse(acc))
+
+    impl(data, Future(start))
+  }
+
+  def toList(implicit executor: ExecutionContext): Future[List[A]] =
+    foldLeft[List[A]](Nil)((list, el) => el :: list) map (_.reverse)
+
+  def takeWhile(p: A => Boolean)(implicit executor: ExecutionContext): AsyncStream[A] =
+    new AsyncStream[A](data map {
+      case None => None
+      case Some(pair) if !p(pair.first) => None
+      case Some(pair) => Some(Pair(pair.first, pair.second.takeWhile(p)))
+    })
+
+  def take(n: Int)(implicit executor: ExecutionContext): AsyncStream[A] =
+    if (n <= 0) nil
+    else AsyncStream(opT(data).map(p => Pair(p.first, p.second.take(n - 1))).run)
+}
+
+object AsyncStream {
+  def nil[A](implicit executor: ExecutionContext): AsyncStream[A] = AsyncStream(None.point[Future])
+  def single[A](item: A)(implicit executor: ExecutionContext): AsyncStream[A] =
+    AsyncStream(Pair(item, nil[A]).some.point[Future])
+
+  def generate[S, A](start: S)(gen: S => Future[Option[(S, A)]])(implicit executor: ExecutionContext): AsyncStream[A] =
+    AsyncStream(opT(gen(start)).map(p => Pair(p._2, generate(p._1)(gen))).run)
+
+  def concat[A](s1: AsyncStream[A], s2: AsyncStream[A])(implicit executor: ExecutionContext): AsyncStream[A] =
+    new AsyncStream[A](s1.data.flatMap {
+      case None => s2.data
+      case Some(p) => Pair(p.first, concat(p.second, s2)).some.point[Future]
+    })
+}

src/main/scala/asyncstreams/AsyncStreamMonad.scala

@@ -0,0 +1,27 @@
+package asyncstreams
+
+import scala.concurrent.{ExecutionContext, Future}
+import scalaz.std.scalaFuture._
+import scalaz.syntax.std.option._
+import scalaz.syntax.monad._
+import scalaz.MonadPlus
+import scalaz.OptionT.{optionT => opT}
+
+class AsyncStreamMonad(implicit executor: ExecutionContext) extends MonadPlus[AsyncStream] {
+  import AsyncStream._
+
+  override def empty[A] = nil[A]
+
+  override def point[A](a: => A): AsyncStream[A] = single(a)
+
+  override def plus[A](a: AsyncStream[A], b: => AsyncStream[A]) = concat(a, b)
+
+  override def bind[A, B](ma: AsyncStream[A])(f: A => AsyncStream[B]): AsyncStream[B] = {
+    val resData = opT(ma.data).flatMap(pair =>
+      opT(f(pair.first).data).map(pair2 =>
+        Pair(pair2.first, concat(pair2.second, bind(pair.second)(f)))
+      )
+    ).run
+    AsyncStream(resData)
+  }
+}

src/main/scala/asyncstreams/Pair.scala

@@ -0,0 +1,10 @@
+package asyncstreams
+
+class Pair[A, B](fp: A, sp: => B) {
+  val first = fp
+  lazy val second = sp
+}
+
+object Pair {
+  def apply[A, B](first:  A, second: => B) = new Pair[A, B](first, second)
+}

src/main/scala/asyncstreams/package.scala

@@ -0,0 +1,10 @@
+import scala.concurrent.{ExecutionContext, Future}
+import scalaz.{Monad, StateT}
+
+package object asyncstreams {
+  type FState[S, A] = StateT[Future, S, A]
+  type Chunk[A, B] = Option[Pair[A, B]]
+
+  implicit def asyncStreamInstance(implicit executor: ExecutionContext): Monad[AsyncStream] =
+    new AsyncStreamMonad
+}

src/test/scala/asyncstreams/AsyncStreamTests.scala

@@ -0,0 +1,60 @@
+package asyncstreams
+
+import scala.concurrent.{Await, Future}
+import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext.Implicits.global
+import asyncstreams.AsyncStream._
+
+import scalaz.std.scalaFuture._
+import scalaz.syntax.std.option._
+import scalaz.syntax.monad._
+import scalaz.syntax.std.boolean._
+
+
+class AsyncStreamTests extends BaseSuite {
+  private def makeStream(l: List[Int]) = generate(l)(l => l.nonEmpty.option((l.tail, l.head)).point[Future])
+
+  private def makeInfStream = generate(0)(v => Future((v + 1, v).some))
+
+  private def wait[T](f: Future[T]): T = Await.result(f, 10.seconds)
+
+  test("foldLeft") {
+    val s2 = makeStream(2 :: 3 :: Nil)
+    val f = s2.foldLeft(List[Int]())((list, el) => el :: list)
+    wait(f) shouldBe List(3, 2)
+  }
+
+  test("concat") {
+    val s1 = makeStream(0 :: 1 :: Nil)
+    val s2 = makeStream(2 :: 3 :: Nil)
+    val f = concat(s1, s2)
+    wait(f.toList) shouldBe List(0, 1, 2, 3)
+  }
+
+  test("working as monad") {
+    val s1 = makeStream(0 :: 1 :: Nil)
+    val s2 = makeStream(2 :: 3 :: Nil)
+
+    val res = for {
+      v1 <- s1
+      v2 <- s2
+    } yield v1 * v2
+
+    wait(res.toList) shouldBe List(0, 0, 2, 3)
+  }
+
+  test("takeWhile") {
+    val r = makeInfStream.takeWhile(_ < 4)
+    wait(r.toList) shouldBe List(0, 1, 2, 3)
+  }
+
+  test("take") {
+    val r = makeInfStream.take(3)
+    wait(r.toList) shouldBe List(0, 1, 2)
+  }
+
+  test("folding large stream should not crash") {
+    val r = makeInfStream.takeWhile(_ < 1000000)
+    wait(r.toList) shouldBe (0 to 999999)
+  }
+}