Merge pull request #2 from danslapman/future-abstraction

Fixes #1

Author: danslapman

build.sbt

@@ -1,12 +1,13 @@
 name := "asyncstreams"
 
-version := "0.4-SNAPSHOT"
+version := "0.4"
 
 scalaVersion := "2.11.8"
 
 addCompilerPlugin("org.spire-math" %% "kind-projector" % "0.9.3")
 
 libraryDependencies ++= Seq(
   "org.scalaz" %% "scalaz-core" % "7.2.8",
+  "com.twitter" %% "util-core" % "6.41.0" % Test,
   "org.scalatest" %% "scalatest" % "3.0.1" % Test
 )

copying.txt

@@ -1,7 +1,7 @@
             DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
                     Version 2, December 2004
 
- Copyright (C) 2016 Daniil Smirnov <[email protected]>
+ Copyright (C) 2016 - 2017 Daniil Smirnov <[email protected]>
 
  Everyone is permitted to copy and distribute verbatim or modified
  copies of this license document, and changing it is allowed as long

src/main/scala/asyncstreams/AsyncStream.scala

@@ -3,74 +3,74 @@ 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
+import scalaz.Monad
+import scalaz.syntax.monad._
 
-case class AsyncStream[A](data: Future[Step[A, AsyncStream[A]]]) {
+case class AsyncStream[F[+_]: Monad, A](data: F[Step[A, AsyncStream[F, A]]]) {
   import AsyncStream._
 
-  def foldLeft[B](start: B)(f: (B, A) => B)(implicit executor: ExecutionContext): Future[B] = {
-    def impl(d: Future[Step[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
+  def foldLeft[B](start: B)(f: (B, A) => B): F[B] = {
+    def impl(d: F[Step[A, AsyncStream[F, A]]], acc: F[B]): F[B] =
       d.flatMap {
         case END => acc
         case step => impl(step.rest.data, acc map (b => f(b, step.value)))
       }
 
-    impl(data, Future(start))
+    impl(data, start.point[F])
   }
 
-  def to[Col[_]](implicit executor: ExecutionContext, cbf: CanBuildFrom[Nothing, A, Col[A @uV]]): Future[Col[A]] =
+  def to[Col[_]](implicit cbf: CanBuildFrom[Nothing, A, Col[A @uV]]): F[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 {
+  def takeWhile(p: A => Boolean): AsyncStream[F, A] =
+    new AsyncStream[F, A](data map {
       case END => END
       case step if !p(step.value) => END
       case step => Step(step.value, step.rest.takeWhile(p))
     })
 
 
-  def take(n: Int)(implicit executor: ExecutionContext): AsyncStream[A] =
+  def take(n: Int): AsyncStream[F, A] =
     if (n <= 0) nil
     else AsyncStream(data.map {
       case END => END
       case p => Step(p.value, p.rest.take(n - 1))
     })
 
-  def foreach[U](f: (A) => U)(implicit executor: ExecutionContext): Future[Unit] =
+  def foreach[U](f: (A) => U): F[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(identity)
+  def foreachF[U](f: (A) => F[U]): F[Unit] =
+    foldLeft(().point[F])((fu: F[Unit], a: A) => fu.flatMap(_ => f(a)).map(_ => ())).flatMap(identity)
 
-  def flatten[B](implicit asIterable: A => GenIterable[B], executor: ExecutionContext): AsyncStream[B] = {
-    val streamChunk = (p: Step[A, AsyncStream[A]]) =>
-      concat(generate(asIterable(p.value))(it => if (it.nonEmpty) Future(it.head, it.tail) else ENDF), p.rest.flatten)
+  def flatten[B](implicit asIterable: A => GenIterable[B]): AsyncStream[F, B] = {
+    val streamChunk = (p: Step[A, AsyncStream[F, A]]) =>
+      concat(generate(asIterable(p.value))(it => if (it.nonEmpty) (it.head, it.tail).point[F] else ENDF[F]), p.rest.flatten)
 
     AsyncStream(data.flatMap {
-      case END => ENDF
+      case END => ENDF[F]
       case step => streamChunk(step).data
     })
   }
 }
 
-
 object AsyncStream {
-  def nil[A](implicit executor: ExecutionContext): AsyncStream[A] = AsyncStream(ENDF)
-  def single[A](item: A)(implicit executor: ExecutionContext): AsyncStream[A] =
-    AsyncStream(Future(Step(item, nil[A])))
+  def nil[F[+_]: Monad, A]: AsyncStream[F, A] = AsyncStream(ENDF[F])
+  def single[F[+_]: Monad, A](item: A): AsyncStream[F, A] =
+    AsyncStream(Step(item, nil[F, A]).point[F])
 
-  def generate[S, A](start: S)(gen: S => Future[(A, S)])(implicit executor: ExecutionContext): AsyncStream[A] =
+  def generate[F[+_]: Monad, S, A](start: S)(gen: S => F[(A, S)]): AsyncStream[F, A] =
     AsyncStream(gen(start).map {
       case END => END
       case (el, rest) => Step(el, generate(rest)(gen))
     })
 
-  def concat[A](s1: AsyncStream[A], s2: AsyncStream[A])(implicit executor: ExecutionContext): AsyncStream[A] =
-    new AsyncStream[A](s1.data.flatMap {
+  def concat[F[+_]: Monad, A](s1: AsyncStream[F, A], s2: AsyncStream[F, A]): AsyncStream[F, A] =
+    new AsyncStream[F, A](s1.data.flatMap {
       case END => s2.data
-      case step => Future(Step(step.value, concat(step.rest, s2)))
+      case step => Step(step.value, concat(step.rest, s2)).point[F]
     })
 }
 

src/main/scala/asyncstreams/AsyncStreamMonad.scala

@@ -1,18 +1,19 @@
 package asyncstreams
 
-import scala.concurrent.{ExecutionContext, Future}
-import scalaz.MonadPlus
+import scala.language.higherKinds
+import scalaz.{Monad, MonadPlus}
+import scalaz.syntax.monad._
 
-class AsyncStreamMonad(implicit executor: ExecutionContext) extends MonadPlus[AsyncStream] {
+class AsyncStreamMonad[F[+_]: Monad] extends MonadPlus[AsyncStream[F, ?]] {
   import AsyncStream._
 
-  override def empty[A] = nil[A]
+  override def empty[A]: AsyncStream[F, A] = nil[F, A]
 
-  override def point[A](a: => A): AsyncStream[A] = single(a)
+  override def point[A](a: => A): AsyncStream[F, A] = single(a)
 
-  override def plus[A](a: AsyncStream[A], b: => AsyncStream[A]) = concat(a, b)
+  override def plus[A](a: AsyncStream[F, A], b: => AsyncStream[F, A]): AsyncStream[F, A] = concat(a, b)
 
-  override def bind[A, B](ma: AsyncStream[A])(f: A => AsyncStream[B]): AsyncStream[B] =
+  override def bind[A, B](ma: AsyncStream[F, A])(f: A => AsyncStream[F, B]): AsyncStream[F, B] =
     AsyncStream(
       ma.data.flatMap {
         case END => ENDF
@@ -23,29 +24,28 @@ class AsyncStreamMonad(implicit executor: ExecutionContext) extends MonadPlus[As
     )
 }
 
-trait AsyncStreamMonadFunctions {
-  def foreach[A, S](stream: AsyncStream[A])(f: A => FState[S, _])
-    (implicit ex: ExecutionContext): FState[S, Unit] =
+class AsyncStreamMonadFunctions[F[+_]: Monad] {
+  def foreach[A, S](stream: AsyncStream[F, A])(f: A => FState[F, S, _]): FState[F, S, Unit] =
     FState(s => {
-      stream.foldLeft(Future(s))((futureS, a) => futureS.flatMap(s2 => f(a)(s2).map(_._2)))
+      stream.foldLeft(s.point[F])((fS, a) => fS.flatMap(s2 => f(a)(s2).map(_._2)))
         .flatMap(identity).map(((), _))
     })
 
-  def isEmpty[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, Boolean] =
+  def isEmpty[A, S](stream: AsyncStream[F, A]): FState[F, S, Boolean] =
     FState(s => stream.data.map(step => (step eq END, s)))
 
-  def isEmpty[A, S : FStateMonad](f: S => AsyncStream[A])(implicit fsm: FStateMonad[S], ex: ExecutionContext): FState[S, Boolean] =
+  def isEmpty[A, S](f: S => AsyncStream[F, A])(implicit fsm: FStateMonad[F, S]): FState[F, S, Boolean] =
     fsm.fcondS((s: S) => isEmpty(f(s)))
 
-  def notEmpty[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, Boolean] =
+  def notEmpty[A, S](stream: AsyncStream[F, A]): FState[F, S, Boolean] =
     FState(s => stream.data map (step => (!(step eq END), s)))
 
-  def notEmpty[A, S](f: S => AsyncStream[A])(implicit fsm: FStateMonad[S], ex: ExecutionContext): FState[S, Boolean] =
+  def notEmpty[A, S](f: S => AsyncStream[F, A])(implicit fsm: FStateMonad[F, S]): FState[F, S, Boolean] =
     fsm.fcondS(s => notEmpty(f(s)))
 
-  def get[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, (A, AsyncStream[A])] =
+  def get[A, S](stream: AsyncStream[F, A]): FState[F, S, (A, AsyncStream[F, A])] =
     FState(s => stream.data.map(step => ((step.value, step.rest), s)))
 
-  def generateS[S,A](start: S)(gen: FState[S, A])(implicit ex: ExecutionContext) =
+  def generateS[S, A](start: S)(gen: FState[F, S, A]): AsyncStream[F, A] =
     AsyncStream.generate(start)(gen.func)
 }

src/main/scala/asyncstreams/FState.scala

@@ -1,36 +1,37 @@
 package asyncstreams
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.language.higherKinds
+import scalaz.Monad
+import scalaz.syntax.monad._
 
-class FState[S, A](val func: S => Future[(A, S)]) {
+class FState[F[+_]: Monad, S, A](val func: S => F[(A, S)]) {
   import FState._
 
-  def apply(s: S) = func(s)
+  def apply(s: S): F[(A, S)] = func(s)
 
-  def flatMap[B](f: A => FState[S, B])(implicit ex: ExecutionContext): FState[S, B] = FState[S, B](
+  def flatMap[B](f: A => FState[F, S, B]): FState[F, S, B] = FState[F, S, B](
     (s: S) => func(s).flatMap {
-      case END => ENDF
+      case END => ENDF[F]
       case (fst, snd) => f(fst)(snd)
     }
   )
 
-  def map[B](f: A => B)(implicit ex: ExecutionContext): FState[S, B] =
-    flatMap((a: A) => FState.unit(f(a)))
+  def map[B](f: A => B): FState[F, S, B] = flatMap((a: A) => FState.unit(f(a)))
 
-  def bind[B](f: A => FState[S, B])(implicit ex: ExecutionContext): FState[S, B] =
+  def bind[B](f: A => FState[F, S, B]): FState[F, S, B] =
     FState((s: S) => func(s) flatMap {
-      case END => ENDF
+      case END => ENDF[F]
       case (fst, snd) => f(fst)(snd)
     })
 
-  def filter(p: A => Boolean)(implicit executor: ExecutionContext): FState[S, A] =
-    bind(a => if (p(a)) unit(a) else empty[S, A])
+  def filter(p: A => Boolean): FState[F, S, A] =
+    bind(a => if (p(a)) unit(a) else empty[F, S, A])
 
-  def withFilter(p: A => Boolean)(implicit executor: ExecutionContext): FState[S, A] = filter(p)
+  def withFilter(p: A => Boolean): FState[F, S, A] = filter(p)
 }
 
 object FState {
-  def apply[S, A](f: S => Future[(A, S)]) = new FState[S, A](f)
-  def unit[S, A](a: => A)(implicit ex: ExecutionContext) = FState[S, A]((s: S) => Future((a, s)))
-  def empty[S, A](implicit ex: ExecutionContext) = FState[S, A]((s: S) => ENDF)
+  def apply[F[+_]: Monad, S, A](f: S => F[(A, S)]) = new FState[F, S, A](f)
+  def unit[F[+_]: Monad, S, A](a: => A) = FState[F, S, A]((s: S) => (a, s).point[F])
+  def empty[F[+_]: Monad, S, A] = FState[F, S, A]((s: S) => ENDF[F])
 }

src/main/scala/asyncstreams/FStateMonad.scala

@@ -1,36 +1,36 @@
 package asyncstreams
 
-import scala.concurrent.{ExecutionContext, Future}
-import scalaz.MonadPlus
+import scala.language.higherKinds
+import scalaz.{Monad, MonadPlus}
+import scalaz.syntax.monad._
 
-class FStateMonad[S](implicit ex: ExecutionContext)
-  extends MonadPlus[FState[S, ?]] with FStateMonadFunctions {
-  type F[X] = FState[S, X]
+class FStateMonad[Fu[+_]: Monad, S] extends FStateMonadFunctions[Fu] with MonadPlus[FState[Fu, S, ?]] {
+  type FS[X] = FState[Fu, S, X]
 
-  override def empty[A]: F[A] = FState.empty[S, A]
+  override def empty[A]: FS[A] = FState.empty[Fu, S, A]
 
-  override def point[A](a: => A): F[A] = FState.unit(a)
+  override def point[A](a: => A): FS[A] = FState.unit(a)
 
-  override def bind[A, B](m: F[A])(f: A => F[B]): F[B] = m.bind(f)
+  override def bind[A, B](m: FS[A])(f: A => FS[B]): FS[B] = m.bind(f)
 
-  override def plus[A](a: F[A],b: => F[A]): F[A] = bind(a)(_ => b)
+  override def plus[A](a: FS[A], b: => FS[A]): FS[A] = bind(a)(_ => b)
 
-  def condS(f: S => Boolean): F[Boolean] = bind(getS[S])(vs => point(f(vs)))
-  def fcondS(f: S => F[Boolean]): F[Boolean] = bind(getS[S])(f)
-  def modS(f: S => S): F[S] = bind(getS[S])(vs => putS(f(vs)))
+  def condS(f: S => Boolean): FS[Boolean] = bind(getS[S])(vs => point(f(vs)))
+  def fcondS(f: S => FS[Boolean]): FS[Boolean] = bind(getS[S])(f)
+  def modS(f: S => S): FS[S] = bind(getS[S])(vs => putS(f(vs)))
 
-  def forM_[A](cond: S => Boolean, mod: S => S)(action: => F[A]): F[Unit] =
+  def forM_[A](cond: S => Boolean, mod: S => S)(action: => FS[A]): FS[Unit] =
     whileM_(condS(cond), bind(action)(va => modS(mod)))
 }
 
-trait FStateMonadFunctions {
-  def getS[S](implicit ex: ExecutionContext): FState[S, S] = FState((s: S) => Future((s, s)))
+class FStateMonadFunctions[Fu[+_]: Monad] {
+  def getS[S]: FState[Fu, S, S] = FState((s: S) => (s, s).point[Fu])
 
-  def putS[S](news: S)(implicit ex: ExecutionContext): FState[S, S] = FState((_: S) => Future((news, news)))
+  def putS[S](news: S): FState[Fu, S, S] = FState((_: S) => (news, news).point[Fu])
 
-  def condS[S](f: S => Boolean)(implicit m: FStateMonad[S]): FStateMonad[S]#F[Boolean] = m.condS(f)
+  def condS[S](f: S => Boolean)(implicit m: FStateMonad[Fu, S]): FStateMonad[Fu, S]#FS[Boolean] = m.condS(f)
 
-  def fconds[S](f: S => FState[S, Boolean])(implicit m: FStateMonad[S]): FStateMonad[S]#F[Boolean] = m.fconds(f)
+  def fconds[S](f: S => FState[Fu, S, Boolean])(implicit m: FStateMonad[Fu, S]): FStateMonad[Fu, S]#FS[Boolean] = m.fconds(f)
 
-  def modS[S](f: S => S)(implicit m: FStateMonad[S]): FStateMonad[S]#F[S] = m.modS(f)
+  def modS[S](f: S => S)(implicit m: FStateMonad[Fu, S]): FStateMonad[Fu, S]#FS[S] = m.modS(f)
 }

src/main/scala/asyncstreams/Step.scala

@@ -1,8 +1,8 @@
 package asyncstreams
 
 class Step[A, B](fp: A, sp: => B) {
-  val value = fp
-  lazy val rest = sp
+  val value: A = fp
+  lazy val rest: B = sp
 }
 
 object Step {

src/main/scala/asyncstreams/package.scala

@@ -1,14 +1,15 @@
-import scala.concurrent.{ExecutionContext, Future}
+import scala.language.higherKinds
 import scalaz.Monad
 
 package object asyncstreams {
   final val END: Null = null
-  final def ENDF(implicit executor: ExecutionContext): Future[Null] = Future(END)
+  final def ENDF[F[+_]: Monad](implicit fm: Monad[F]): F[Null] = fm.point(END)
 
-  implicit def asyncStreamInstance(implicit executor: ExecutionContext): Monad[AsyncStream] =
-    new AsyncStreamMonad
+  implicit def asyncStreamInstance[F[+_]: Monad]: Monad[AsyncStream[F, ?]] = new AsyncStreamMonad[F]
 
-  def fStateInstance[S](implicit executor: ExecutionContext) = new FStateMonad[S]
+  def fStateInstance[F[+_]: Monad, S] = new FStateMonad[F, S]
 
-  object monadops extends FStateMonadFunctions with AsyncStreamMonadFunctions
+  def fStateOps[F[+_]: Monad] = new FStateMonadFunctions[F]
+
+  def streamOps[F[+_]: Monad] = new AsyncStreamMonadFunctions[F]
 }

src/test/scala/asyncstreams/AsyncStreamMonadicOperationsTests.scala renamed to src/test/scala/asyncstreams/scalaFutures/AsyncStreamMonadicOperationsTests.scala

@@ -1,19 +1,26 @@
-package asyncstreams
+package asyncstreams.scalaFutures
 
-import monadops._
-import AsyncStream._
+import asyncstreams.AsyncStream.generate
+import asyncstreams.{streamOps, fStateOps}
+import asyncstreams.{AsyncStream, BaseSuite, ENDF, fStateInstance}
 
+import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration._
 import scala.concurrent.{Await, Future}
-import scala.concurrent.ExecutionContext.Implicits.global
+import scalaz.std.scalaFuture._
 
 class AsyncStreamMonadicOperationsTests extends BaseSuite {
-  def makeStream(l: List[Int]) = generate(l)(l => if (l.isEmpty) ENDF else Future((l.head, l.tail)))
+  private val so = streamOps[Future]
+  import so._
+  private val fso = fStateOps[Future]
+  import fso._
+
+  def makeStream(l: List[Int]) = generate(l)(l => if (l.isEmpty) ENDF[Future] else Future((l.head, l.tail)))
 
   private def wait[T](f: Future[T]): T = Await.result(f, 10.seconds)
 
   test("foreach") {
-    implicit val fsm = fStateInstance[Int]
+    implicit val fsm = fStateInstance[Future, Int]
 
     val fstate = for {
       _ <- foreach(makeStream(0 :: 1 :: 2 :: Nil)) {
@@ -26,8 +33,8 @@ class AsyncStreamMonadicOperationsTests extends BaseSuite {
   }
 
   test("get, isEmpty") {
-    case class State(counter: Int, stream: AsyncStream[Int])
-    implicit val fsm = fStateInstance[State]
+    case class State(counter: Int, stream: AsyncStream[Future, Int])
+    implicit val fsm = fStateInstance[Future, State]
 
     val stream = makeStream(0 :: 1 :: 2 :: 3 :: Nil)
 
@@ -44,7 +51,7 @@ class AsyncStreamMonadicOperationsTests extends BaseSuite {
   }
 
   test("FState as generator") {
-    implicit val fsm = fStateInstance[Int]
+    implicit val fsm = fStateInstance[Future, Int]
 
     val stream = generateS(0) {
       for {
@@ -57,7 +64,7 @@ class AsyncStreamMonadicOperationsTests extends BaseSuite {
   }
 
   test("Generate finite stream") {
-    implicit val fsm = fStateInstance[Int]
+    implicit val fsm = fStateInstance[Future, Int]
 
     val stream = generateS(0) {
       for {