Add StateT based runtime (#24)

Author: notxcain

core/src/main/scala/aecor/aggregate/Folder.scala

@@ -1,8 +1,12 @@
 package aecor.aggregate
 
+import cats.{ Foldable, Monad }
+import cats.implicits._
+
 trait Folder[F[_], A, B] {
   def zero: B
   def fold(b: B, a: A): F[B]
+  def consume[I[_]: Foldable](f: I[A])(implicit F: Monad[F]): F[B] = f.foldM(zero)(fold)
 }
 
 object Folder {
@@ -11,5 +15,4 @@ object Folder {
       override def zero: B = b
       override def fold(b: B, a: A): F[B] = f(b)(a)
     }
-
 }

core/src/main/scala/aecor/aggregate/StateRuntime.scala

@@ -0,0 +1,51 @@
+package aecor.aggregate
+
+import aecor.data.Handler
+import cats.data._
+import cats.implicits._
+import cats.{ Monad, ~> }
+
+object StateRuntime {
+
+  /**
+    * Creates an aggregate runtime that uses StateT as a target context
+    *
+    * This runtime doesn't account for correlation,
+    * i.e. all operations are executed against common sequence of events
+    *
+    */
+  def shared[Op[_], S, E, F[_]: Monad](
+    behavior: Op ~> Handler[S, E, ?]
+  )(implicit folder: Folder[F, E, S]): Op ~> StateT[F, Vector[E], ?] =
+    new (Op ~> StateT[F, Vector[E], ?]) {
+      override def apply[A](fa: Op[A]): StateT[F, Vector[E], A] =
+        for {
+          events <- StateT.get[F, Vector[E]]
+          state <- StateT.lift(folder.consume(events))
+          result <- {
+            val (es, r) = behavior(fa).run(state)
+            StateT.modify[F, Vector[E]](_ ++ es).map(_ => r)
+          }
+        } yield result
+    }
+
+  /**
+    * Creates an aggregate runtime that uses StateT as a target context
+    *
+    * This runtime uses correlation function to get entity identifier
+    * that is used to execute commands against corresponding
+    * sequence of events
+    *
+    */
+  def correlated[O[_], S, E, F[_]: Monad](
+    behavior: O ~> Handler[S, E, ?],
+    correlation: Correlation[O]
+  )(implicit folder: Folder[F, E, S]): O ~> StateT[F, Map[String, Vector[E]], ?] =
+    new (O ~> StateT[F, Map[String, Vector[E]], ?]) {
+      override def apply[A](fa: O[A]): StateT[F, Map[String, Vector[E]], A] = {
+        val inner: O ~> StateT[F, Vector[E], ?] = shared(behavior)
+        val entityId = correlation(fa)
+        inner(fa).transformS(_.getOrElse(entityId, Vector.empty[E]), _.updated(entityId, _))
+      }
+    }
+}

core/src/main/scala/aecor/data/Folded.scala

@@ -1,6 +1,21 @@
 package aecor.data
 
 import aecor.data.Folded.{ Impossible, Next }
+import cats.kernel.Eq
+import cats.{
+  Alternative,
+  Applicative,
+  CoflatMap,
+  Eval,
+  Monad,
+  MonadCombine,
+  MonadError,
+  Now,
+  Show,
+  TraverseFilter
+}
+
+import scala.annotation.tailrec
 
 sealed abstract class Folded[+A] extends Product with Serializable {
   def fold[B](impossible: => B, next: A => B): B = this match {
@@ -19,10 +34,25 @@ sealed abstract class Folded[+A] extends Product with Serializable {
     case Impossible => that
     case Next(a) => a
   }
+  def orElse[AA >: A](that: Folded[AA]): Folded[AA] = this match {
+    case Next(_) => this
+    case Impossible => that
+  }
+  def isNext: Boolean = fold(false, _ => true)
+  def isImpossible: Boolean = !isNext
+
+  def filter(f: A => Boolean): Folded[A] = this match {
+    case Next(a) if f(a) => this
+    case _ => Impossible
+  }
+  def exists(f: A => Boolean): Boolean = filter(f).isNext
+  def forall(f: A => Boolean): Boolean = fold(true, f)
+
+  def toOption: Option[A] = fold(None, Some(_))
 }
-object Folded {
-  private final case object Impossible extends Folded[Nothing]
-  private final case class Next[+A](a: A) extends Folded[A]
+object Folded extends FoldedInstances {
+  final case object Impossible extends Folded[Nothing]
+  final case class Next[+A](a: A) extends Folded[A]
   def impossible[A]: Folded[A] = Impossible
   def next[A](a: A): Folded[A] = Next(a)
   object syntax {
@@ -32,3 +62,107 @@ object Folded {
     def impossible[A]: Folded[A] = Folded.impossible
   }
 }
+
+trait FoldedInstances {
+  implicit val aecorDataInstancesForFolded
+    : TraverseFilter[Folded] with MonadError[Folded, Unit] with MonadCombine[Folded] with Monad[
+      Folded
+    ] with CoflatMap[Folded] with Alternative[Folded] =
+    new TraverseFilter[Folded] with MonadError[Folded, Unit] with MonadCombine[Folded]
+    with Monad[Folded] with CoflatMap[Folded] with Alternative[Folded] {
+
+      def empty[A]: Folded[A] = Impossible
+
+      def combineK[A](x: Folded[A], y: Folded[A]): Folded[A] = x orElse y
+
+      def pure[A](x: A): Folded[A] = Next(x)
+
+      override def map[A, B](fa: Folded[A])(f: A => B): Folded[B] =
+        fa.map(f)
+
+      def flatMap[A, B](fa: Folded[A])(f: A => Folded[B]): Folded[B] =
+        fa.flatMap(f)
+
+      @tailrec
+      def tailRecM[A, B](a: A)(f: A => Folded[Either[A, B]]): Folded[B] =
+        f(a) match {
+          case Impossible => Impossible
+          case Next(Left(a1)) => tailRecM(a1)(f)
+          case Next(Right(b)) => Next(b)
+        }
+
+      override def map2[A, B, Z](fa: Folded[A], fb: Folded[B])(f: (A, B) => Z): Folded[Z] =
+        fa.flatMap(a => fb.map(b => f(a, b)))
+
+      override def map2Eval[A, B, Z](fa: Folded[A],
+                                     fb: Eval[Folded[B]])(f: (A, B) => Z): Eval[Folded[Z]] =
+        fa match {
+          case Impossible => Now(Impossible)
+          case Next(a) => fb.map(_.map(f(a, _)))
+        }
+
+      def coflatMap[A, B](fa: Folded[A])(f: Folded[A] => B): Folded[B] =
+        if (fa.isNext) Next(f(fa)) else Impossible
+
+      def foldLeft[A, B](fa: Folded[A], b: B)(f: (B, A) => B): B =
+        fa match {
+          case Impossible => b
+          case Next(a) => f(b, a)
+        }
+
+      def foldRight[A, B](fa: Folded[A], lb: Eval[B])(f: (A, Eval[B]) => Eval[B]): Eval[B] =
+        fa match {
+          case Impossible => lb
+          case Next(a) => f(a, lb)
+        }
+
+      def raiseError[A](e: Unit): Folded[A] = Impossible
+
+      def handleErrorWith[A](fa: Folded[A])(f: (Unit) => Folded[A]): Folded[A] = fa orElse f(())
+
+      def traverseFilter[G[_], A, B](
+        fa: Folded[A]
+      )(f: A => G[Option[B]])(implicit G: Applicative[G]): G[Folded[B]] =
+        fa match {
+          case Impossible => G.pure(Impossible)
+          case Next(a) =>
+            G.map(f(a)) {
+              case Some(aa) => Next(aa)
+              case None => Impossible
+            }
+        }
+
+      override def traverse[G[_]: Applicative, A, B](fa: Folded[A])(f: A => G[B]): G[Folded[B]] =
+        fa match {
+          case Impossible => Applicative[G].pure(Impossible)
+          case Next(a) => Applicative[G].map(f(a))(Next(_))
+        }
+
+      override def filter[A](fa: Folded[A])(p: A => Boolean): Folded[A] =
+        fa.filter(p)
+
+      override def exists[A](fa: Folded[A])(p: A => Boolean): Boolean =
+        fa.exists(p)
+
+      override def forall[A](fa: Folded[A])(p: A => Boolean): Boolean =
+        fa.forall(p)
+
+      override def isEmpty[A](fa: Folded[A]): Boolean =
+        fa.isImpossible
+    }
+
+  implicit def aecorDataShowForFolded[A](implicit A: Show[A]): Show[Folded[A]] =
+    new Show[Folded[A]] {
+      def show(fa: Folded[A]): String = fa match {
+        case Next(a) => s"Next(${A.show(a)})"
+        case Impossible => "Impossible"
+      }
+    }
+
+  implicit def aecorDataEqForFolded[A](implicit A: Eq[A]): Eq[Folded[A]] =
+    Eq.instance {
+      case (Next(l), Next(r)) => A.eqv(l, r)
+      case (Impossible, Impossible) => true
+      case _ => false
+    }
+}

tests/src/test/scala/aecor/tests/FoldedTests.scala

@@ -0,0 +1,94 @@
+package aecor.tests
+
+import aecor.data.Folded
+import cats.{ Cartesian, CoflatMap, Eval, Later, Monad, MonadCombine, MonadError, TraverseFilter }
+import cats.laws.{ ApplicativeLaws, CoflatMapLaws, FlatMapLaws, MonadLaws }
+import cats.laws.discipline._
+import Folded.syntax._
+import org.scalacheck.{ Arbitrary, Cogen }
+
+class FoldedTests extends LawSuite {
+
+  implicit def arbitraryFolded[A](implicit A: Arbitrary[Option[A]]): Arbitrary[Folded[A]] =
+    Arbitrary(A.arbitrary.map(_.map(_.next).getOrElse(impossible)))
+
+  implicit def cogenFolded[A](implicit A: Cogen[Option[A]]): Cogen[Folded[A]] =
+    A.contramap(_.toOption)
+
+  checkAll("Folded[Int]", CartesianTests[Folded].cartesian[Int, Int, Int])
+  checkAll("Cartesian[Folded]", SerializableTests.serializable(Cartesian[Folded]))
+
+  checkAll("Folded[Int]", CoflatMapTests[Folded].coflatMap[Int, Int, Int])
+  checkAll("CoflatMap[Folded]", SerializableTests.serializable(CoflatMap[Folded]))
+
+  checkAll("Folded[Int]", MonadCombineTests[Folded].monadCombine[Int, Int, Int])
+  checkAll("MonadCombine[Folded]", SerializableTests.serializable(MonadCombine[Folded]))
+
+  checkAll("Folded[Int]", MonadTests[Folded].monad[Int, Int, Int])
+  checkAll("Monad[Folded]", SerializableTests.serializable(Monad[Folded]))
+
+  checkAll(
+    "Folded[Int] with Folded",
+    TraverseFilterTests[Folded].traverseFilter[Int, Int, Int, Int, Folded, Folded]
+  )
+  checkAll("TraverseFilter[Folded]", SerializableTests.serializable(TraverseFilter[Folded]))
+
+  checkAll("Folded with Unit", MonadErrorTests[Folded, Unit].monadError[Int, Int, Int])
+  checkAll("MonadError[Folded, Unit]", SerializableTests.serializable(MonadError[Folded, Unit]))
+
+  test("show") {
+    impossible[Int].show should ===("Impossible")
+    1.next.show should ===("Next(1)")
+
+    forAll { fs: Folded[String] =>
+      fs.show should ===(fs.toString)
+    }
+  }
+
+  // The following tests check laws which are a different formulation of
+  // laws that are checked. Since these laws are more or less duplicates of
+  // existing laws, we don't check them for all types that have the relevant
+  // instances.
+
+  test("Kleisli associativity") {
+    forAll { (l: Long, f: Long => Folded[Int], g: Int => Folded[Char], h: Char => Folded[String]) =>
+      val isEq = FlatMapLaws[Folded].kleisliAssociativity(f, g, h, l)
+      isEq.lhs should ===(isEq.rhs)
+    }
+  }
+
+  test("Cokleisli associativity") {
+    forAll { (l: Folded[Long], f: Folded[Long] => Int, g: Folded[Int] => Char, h: Folded[Char] => String) =>
+      val isEq = CoflatMapLaws[Folded].cokleisliAssociativity(f, g, h, l)
+      isEq.lhs should ===(isEq.rhs)
+    }
+  }
+
+  test("applicative composition") {
+    forAll { (fa: Folded[Int], fab: Folded[Int => Long], fbc: Folded[Long => Char]) =>
+      val isEq = ApplicativeLaws[Folded].applicativeComposition(fa, fab, fbc)
+      isEq.lhs should ===(isEq.rhs)
+    }
+  }
+
+  val monadLaws = MonadLaws[Folded]
+
+  test("Kleisli left identity") {
+    forAll { (a: Int, f: Int => Folded[Long]) =>
+      val isEq = monadLaws.kleisliLeftIdentity(a, f)
+      isEq.lhs should ===(isEq.rhs)
+    }
+  }
+
+  test("Kleisli right identity") {
+    forAll { (a: Int, f: Int => Folded[Long]) =>
+      val isEq = monadLaws.kleisliRightIdentity(a, f)
+      isEq.lhs should ===(isEq.rhs)
+    }
+  }
+
+  test("map2Eval is lazy") {
+    val bomb: Eval[Folded[Int]] = Later(sys.error("boom"))
+    impossible[Int].map2Eval(bomb)(_ + _).value should ===(impossible[Int])
+  }
+}