Add AsyncStreamMonadFunctions

Author: danslapman

src/main/scala/asyncstreams/AsyncStream.scala

@@ -11,7 +11,7 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
   def foldLeft[B](start: B)(f: (B, A) => B)(implicit executor: ExecutionContext): Future[B] = {
     def impl(d: Future[Pair[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
       d.flatMap {
-        case null => acc
+        case END => acc
         case pair => impl(pair.second.data, acc map (b => f(b, pair.first)))
       }
 
@@ -24,7 +24,7 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
 
   def takeWhile(p: A => Boolean)(implicit executor: ExecutionContext): AsyncStream[A] =
     new AsyncStream[A](data map {
-      case null => END
+      case END => END
       case pair if !p(pair.first) => END
       case pair => Pair(pair.first, pair.second.takeWhile(p))
     })
@@ -33,7 +33,7 @@ case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]]) {
   def take(n: Int)(implicit executor: ExecutionContext): AsyncStream[A] =
     if (n <= 0) nil
     else AsyncStream(data.map {
-      case null => END
+      case END => END
       case p => Pair(p.first, p.second.take(n - 1))
     })
 }
@@ -46,13 +46,13 @@ object AsyncStream {
 
   def generate[S, A](start: S)(gen: S => Future[(A, S)])(implicit executor: ExecutionContext): AsyncStream[A] =
     AsyncStream(gen(start).map {
-      case null => END
+      case END => 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 null => s2.data
+      case END => s2.data
       case p => Future(Pair(p.first, concat(p.second, s2)))
     })
 }

src/main/scala/asyncstreams/AsyncStreamMonad.scala

@@ -15,10 +15,37 @@ class AsyncStreamMonad(implicit executor: ExecutionContext) extends MonadPlus[As
   override def bind[A, B](ma: AsyncStream[A])(f: A => AsyncStream[B]): AsyncStream[B] =
     AsyncStream(
       ma.data.flatMap {
-        case null => Future(null)
+        case END => ENDF
         case pair => f(pair.first).data.map { pair2 =>
           Pair(pair2.first, concat(pair2.second, bind(pair.second)(f)))
         }
       }
     )
 }
+
+trait AsyncStreamMonadFunctions {
+  def foreach[A, S](stream: AsyncStream[A])(f: A => FState[S, Any])
+    (implicit ex: ExecutionContext): FState[S, Unit] =
+    FState(s => {
+      stream.foldLeft(Future(s))((futureS, a) => futureS.flatMap(s2 => f(a)(s2).map(_._2)))
+        .flatMap(f => f).map(((), _))
+    })
+
+  def isEmpty[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, Boolean] =
+    FState(s => stream.data.map(pair => (pair eq END, s)))
+
+  def isEmpty[A, S : FStateMonad](f: S => AsyncStream[A])(implicit fsm: FStateMonad[S], ex: ExecutionContext): FState[S, Boolean] =
+    fsm.fconds((s: S) => isEmpty(f(s)))
+
+  def notEmpty[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, Boolean] =
+    FState(s => stream.data map (pair => (!(pair eq END), s)))
+
+  def notEmpty[A, S](f: S => AsyncStream[A])(implicit fsm: FStateMonad[S], ex: ExecutionContext): FState[S, Boolean] =
+    fsm.fconds(s => notEmpty(f(s)))
+
+  def get[A, S](stream: AsyncStream[A])(implicit ex: ExecutionContext): FState[S, (A, AsyncStream[A])] =
+    FState(s => stream.data.map(pair => ((pair.first, pair.second), s)))
+
+  def generate[S,A](start: S)(gen: FState[S, A])(implicit ex: ExecutionContext) =
+    AsyncStream.generate(start)(gen.func)
+}

src/main/scala/asyncstreams/FState.scala

@@ -2,7 +2,7 @@ package asyncstreams
 
 import scala.concurrent.{ExecutionContext, Future}
 
-class FState[S, A](func: S => Future[(A, S)]) extends ((S) => Future[(A, S)]) {
+class FState[S, A](val func: S => Future[(A, S)]) extends ((S) => Future[(A, S)]) {
   def apply(s: S) = func(s)
 
   def flatMap[B](f: A => FState[S, B])(implicit ex: ExecutionContext): FState[S, B] = FState[S, B](

src/main/scala/asyncstreams/FStateMonad.scala

@@ -13,7 +13,7 @@ class FStateMonad[S](implicit ex: ExecutionContext)
 
   override def bind[A, B](m: F[A])(f: A => F[B]): F[B] =
     FState((s: S) => m(s) flatMap {
-      case null => ENDF
+      case END => ENDF
       case (fst, snd) => f(fst)(snd)
     })
 
@@ -31,12 +31,14 @@ trait FStateMonadFunctions {
   def gets[S](implicit ex: ExecutionContext): FState[S, S] = FState((s: S) => Future((s, s)))
   def puts[S](news: S)(implicit ex: ExecutionContext): FState[S, S] = FState((_: S) => Future((news, news)))
 
-  def conds[S](f: S => Boolean)(implicit m: FStateMonad[S]): FState[S, Boolean] =
+  def conds[S](f: S => Boolean)(implicit m: FStateMonad[S]): FStateMonad[S]#F[Boolean] =
     m.conds(f)
 
-  def fconds[S](f: S => FState[S, Boolean])(implicit m: FStateMonad[S]): FState[S, Boolean] =
+  /*
+  def fconds[S](f: S => FState[S, Boolean])(implicit m: FStateMonad[S]): FStateMonad[S]#F[Boolean] =
     m.fconds(f)
+    */
 
-  def mods[S : FStateMonad](f: S => S)(implicit m: FStateMonad[S]): FState[S, S] =
+  def mods[S : FStateMonad](f: S => S)(implicit m: FStateMonad[S]): FStateMonad[S]#F[S] =
     m.mods(f)
-}
+}

src/main/scala/asyncstreams/package.scala

@@ -10,5 +10,5 @@ package object asyncstreams {
 
   def fStateInstance[S](implicit executor: ExecutionContext) = new FStateMonad[S]
 
-  object monadops extends FStateMonadFunctions
+  object monadops extends FStateMonadFunctions with AsyncStreamMonadFunctions
 }