Use null as an "EndOfStream" indicator, reimplement AsyncStream without monad transformers

Author: danslapman

src/main/scala/asyncstreams/AsyncStream.scala

@@ -1,50 +1,61 @@
 package asyncstreams
 
+import scala.annotation.unchecked.{uncheckedVariance => uV}
 import scala.collection.generic.CanBuildFrom
-import scala.annotation.unchecked.{ uncheckedVariance => uV }
 import scala.concurrent.{ExecutionContext, Future}
 import scala.language.higherKinds
 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]]]) {
+case class AsyncStream[A](data: Future[Pair[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))
+    def impl(d: Future[Pair[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
+      d.flatMap {
+        case null => acc
+        case pair => impl(pair.second.data, acc map (b => f(b, pair.first)))
+      }
 
     impl(data, Future(start))
   }
 
   def to[Col[_]](implicit executor: ExecutionContext, cbf: CanBuildFrom[Nothing, A, Col[A @uV]]): Future[Col[A]] =
     foldLeft(cbf())((col, el) => col += el).map(_.result())
 
+
   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)))
+      case null => END
+      case pair if !p(pair.first) => END
+      case pair => 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)
+    else AsyncStream(data.map {
+      case null => END
+      case p => Pair(p.first, p.second.take(n - 1))
+    })
 }
 
+
 object AsyncStream {
-  def nil[A](implicit executor: ExecutionContext): AsyncStream[A] = AsyncStream(None.point[Future])
+  def nil[A](implicit executor: ExecutionContext): AsyncStream[A] = AsyncStream(ENDF)
   def single[A](item: A)(implicit executor: ExecutionContext): AsyncStream[A] =
-    AsyncStream(Pair(item, nil[A]).some.point[Future])
+    AsyncStream(Pair(item, nil[A]).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 generate[S, A](start: S)(gen: S => Future[(A, S)])(implicit executor: ExecutionContext): AsyncStream[A] =
+    AsyncStream(gen(start).map {
+      case null => END
+      case (el, rest) => Pair(el, generate(rest)(gen))
+    })
 
   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]
+      case null => s2.data
+      case p => Pair(p.first, concat(p.second, s2)).point[Future]
     })
 }
+

src/main/scala/asyncstreams/AsyncStreamMonad.scala

@@ -1,11 +1,7 @@
 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._
@@ -16,12 +12,13 @@ class AsyncStreamMonad(implicit executor: ExecutionContext) extends MonadPlus[As
 
   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)
-  }
+  override def bind[A, B](ma: AsyncStream[A])(f: A => AsyncStream[B]): AsyncStream[B] =
+    AsyncStream(
+      ma.data.flatMap {
+        case null => Future(null)
+        case pair => f(pair.first).data.map { pair2 =>
+          Pair(pair2.first, concat(pair2.second, bind(pair.second)(f)))
+        }
+      }
+    )
 }

src/main/scala/asyncstreams/package.scala

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

src/test/scala/asyncstreams/AsyncStreamTests.scala

@@ -6,15 +6,14 @@ 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 makeStream(l: List[Int]) = generate(l)(l => ((l.nonEmpty)?(l.head, l.tail)|END).point[Future])
 
-  private def makeInfStream = generate(0)(v => Future((v + 1, v).some))
+  private def makeInfStream = generate(0)(v => Future((v, v + 1)))
 
   private def wait[T](f: Future[T]): T = Await.result(f, 10.seconds)