FState, monadic operations on FState

Author: danslapman

src/main/scala/asyncstreams/FState.scala

@@ -0,0 +1,19 @@
+package asyncstreams
+
+import scala.concurrent.{ExecutionContext, Future}
+
+class FState[S, A](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](
+    (s: S) => func(s) flatMap ((fst: A, snd: S) => f(fst)(snd)).tupled
+  )
+
+  def map[B](f: A => B)(implicit ex: ExecutionContext): FState[S, B] =
+    flatMap((a: A) => FState.unit(f(a)))
+}
+
+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)))
+}

src/main/scala/asyncstreams/FStateMonad.scala

@@ -0,0 +1,42 @@
+package asyncstreams
+
+import scala.concurrent.{ExecutionContext, Future}
+import scalaz.MonadPlus
+
+class FStateMonad[S](implicit ex: ExecutionContext)
+  extends MonadPlus[({ type f[X] = FState[S, X]})#f] with FStateMonadFunctions {
+  type F[X] = FState[S, X]
+
+  override def empty[A]: F[A] = FState((s: S) => ENDF)
+
+  override def point[A](a: => A): F[A] = FState.unit(a)
+
+  override def bind[A, B](m: F[A])(f: A => F[B]): F[B] =
+    FState((s: S) => m(s) flatMap {
+      case null => ENDF
+      case (fst, snd) => f(fst)(snd)
+    })
+
+  override def plus[A](a: F[A],b: => F[A]): F[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 forM_[A](cond: S => Boolean, mod: S => S)(action: => F[A]): F[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)))
+  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] =
+    m.conds(f)
+
+  def fconds[S](f: S => FState[S, Boolean])(implicit m: FStateMonad[S]): FState[S, Boolean] =
+    m.fconds(f)
+
+  def mods[S : FStateMonad](f: S => S)(implicit m: FStateMonad[S]): FState[S, S] =
+    m.mods(f)
+}

src/main/scala/asyncstreams/package.scala

@@ -7,4 +7,8 @@ package object asyncstreams {
 
   implicit def asyncStreamInstance(implicit executor: ExecutionContext): Monad[AsyncStream] =
     new AsyncStreamMonad
+
+  def fStateInstance[S](implicit executor: ExecutionContext) = new FStateMonad[S]
+
+  object monadops extends FStateMonadFunctions
 }

src/test/scala/asyncstreams/FStateTests.scala

@@ -0,0 +1,60 @@
+package asyncstreams
+
+import monadops._
+
+import scala.concurrent.{Await, Future}
+import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext.Implicits.global
+
+class FStateTests extends BaseSuite {
+  private def wait[T](f: Future[T]): T = Await.result(f, 10.seconds)
+
+  test("FState in for-comprehensions") {
+    val fsm = fStateInstance[Int]
+
+    val t = for {
+      a <- fsm.point(10)
+      b = a + 1
+    } yield b
+
+    wait(t(0)) shouldBe (11, 0)
+  }
+
+  test("gets & puts") {
+    val fsm = fStateInstance[Int]
+
+    val t = for {
+      _ <- fsm.whileM_(gets[Int] map (_ < 10), for {
+        i <- gets[Int]
+        _ <- puts(i + 1)
+      } yield (()))
+      v1 <- gets[Int]
+    } yield v1
+
+    wait(t(0)) shouldBe (10, 10)
+  }
+
+  test("conds & mods") {
+    implicit val fsm = fStateInstance[Int]
+
+    val t = for {
+      _ <- fsm.whileM_(conds(_ < 10), mods(_ + 1))
+      v1 <- gets[Int]
+    } yield v1
+
+    wait(t(0)) shouldBe (10, 10)
+  }
+
+  test("forM_") {
+    val fsm = fStateInstance[Int]
+
+    val t = for {
+      _ <- fsm.forM_(_ < 10, _ + 1) {
+        fsm.point("AAAAAA")
+      }
+      v1 <- gets[Int]
+    } yield v1
+
+    wait(t(0)) shouldBe (10, 10)
+  }
+}