Behavior overhaul (#11)

Author: notxcain

.scalafmt.conf

@@ -0,0 +1,9 @@
+danglingParentheses = true
+maxColumn = 100
+rewrite.rules = [PreferCurlyFors, RedundantBraces]
+style = IntelliJ
+align.arrowEnumeratorGenerator = true
+align.openParenCallSite = false
+spaces {
+  inImportCurlyBraces = true
+}

README.md

@@ -1,7 +1,7 @@
 
 [![Build Status](https://img.shields.io/travis/notxcain/aecor/master.svg)](https://travis-ci.org/notxcain/aecor)
 [![Maven Central](https://img.shields.io/maven-central/v/io.aecor/aecor-core_2.11.svg)](https://github.com/notxcain/aecor)
-[![Join the chat at https://gitter.im/notxcain/akka-ddd](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/notxcain/aecor)
+[![Join the chat at https://gitter.im/notxcain/aecor](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/notxcain/aecor)
 
 
 # Aecor

build.sbt

@@ -7,15 +7,9 @@ lazy val buildSettings = Seq(
   crossScalaVersions := Seq("2.11.8", "2.12.0")
 )
 
-lazy val circeVersion = "0.6.1"
-lazy val akkaVersion = "2.4.14"
-lazy val akkaHttpVersion = "10.0.0"
-lazy val reactiveKafkaVersion = "0.13"
-lazy val akkaPersistenceCassandra = "0.21"
-lazy val catsVersion = "0.8.1"
-lazy val akkaHttpJsonVersion = "1.11.0"
-lazy val freekVersion = "0.6.5"
-lazy val kryoSerializationVersion = "0.5.1"
+lazy val akkaVersion = "2.4.16"
+lazy val akkaPersistenceCassandra = "0.22"
+lazy val catsVersion = "0.9.0"
 lazy val logbackVersion = "1.1.7"
 
 lazy val scalaCheckVersion = "1.13.4"
@@ -27,14 +21,9 @@ lazy val paradiseVersion = "2.1.0"
 
 lazy val commonSettings = Seq(
     scalacOptions ++= commonScalacOptions,
-    resolvers ++= Seq(
-      Resolver.bintrayRepo("projectseptemberinc", "maven")
-    ),
     libraryDependencies ++= Seq(
-      compilerPlugin(
-        "org.spire-math" %% "kind-projector" % kindProjectorVersion),
-      compilerPlugin(
-        "org.scalamacros" % "paradise" % paradiseVersion cross CrossVersion.full)
+      compilerPlugin("org.spire-math" %% "kind-projector" % kindProjectorVersion),
+      compilerPlugin("org.scalamacros" % "paradise" % paradiseVersion cross CrossVersion.full)
     ),
     parallelExecution in Test := false,
     scalacOptions in (Compile, doc) := (scalacOptions in (Compile, doc)).value
@@ -49,14 +38,10 @@ lazy val aecor = project
   .settings(aecorSettings)
   .settings(noPublishSettings)
   .aggregate(core, example, schedule, tests)
-  .dependsOn(core,
-             example % "compile-internal",
-             tests % "test-internal -> test")
+  .dependsOn(core, example % "compile-internal", tests % "test-internal -> test")
 
-lazy val core = project
-  .settings(moduleName := "aecor-core")
-  .settings(aecorSettings)
-  .settings(coreSettings)
+lazy val core =
+  project.settings(moduleName := "aecor-core").settings(aecorSettings).settings(coreSettings)
 
 lazy val schedule = project
   .dependsOn(core)
@@ -85,26 +70,33 @@ lazy val coreSettings = Seq(
     "com.typesafe.akka" %% "akka-persistence-query-experimental" % akkaVersion,
     "com.typesafe.akka" %% "akka-slf4j" % akkaVersion,
     "com.typesafe.akka" %% "akka-persistence-cassandra" % akkaPersistenceCassandra,
-    "com.typesafe.akka" %% "akka-stream-kafka" % reactiveKafkaVersion,
-    "ch.qos.logback" % "logback-classic" % logbackVersion,
     "com.chuusai" %% "shapeless" % shapelessVersion,
     "org.typelevel" %% "cats" % catsVersion
   )
 )
 
 lazy val scheduleSettings = commonProtobufSettings
 
-lazy val exampleSettings = Seq(
-  libraryDependencies ++= Seq(
-    "com.typesafe.akka" %% "akka-http" % akkaHttpVersion,
-    "de.heikoseeberger" %% "akka-http-circe" % akkaHttpJsonVersion,
-    ("com.projectseptember" %% "freek" % freekVersion)
-      .exclude("org.typelevel", "cats-free_2.12.0-RC2"),
-    "io.circe" %% "circe-core" % circeVersion,
-    "io.circe" %% "circe-generic" % circeVersion,
-    "io.circe" %% "circe-parser" % circeVersion
+lazy val exampleSettings = {
+  val circeVersion = "0.6.1"
+  val akkaHttpVersion = "10.0.3"
+  val akkaHttpJsonVersion = "1.11.0"
+  val freekVersion = "0.6.5"
+  Seq(
+    resolvers ++= Seq(Resolver.bintrayRepo("projectseptemberinc", "maven")),
+    libraryDependencies ++=
+      Seq(
+        "com.typesafe.akka" %% "akka-http" % akkaHttpVersion,
+        "de.heikoseeberger" %% "akka-http-circe" % akkaHttpJsonVersion,
+        ("com.projectseptember" %% "freek" % freekVersion)
+          .exclude("org.typelevel", "cats-free_2.12.0-RC2"),
+        "io.circe" %% "circe-core" % circeVersion,
+        "io.circe" %% "circe-generic" % circeVersion,
+        "io.circe" %% "circe-parser" % circeVersion,
+        "ch.qos.logback" % "logback-classic" % logbackVersion
+      )
   )
-)
+}
 
 lazy val testingSettings = Seq(
   libraryDependencies ++= Seq(
@@ -142,18 +134,11 @@ lazy val commonScalacOptions = Seq(
   "-Ypartial-unification"
 )
 
-lazy val warnUnusedImport = Seq(
-  scalacOptions in (Compile, console) ~= {
-    _.filterNot(Set("-Ywarn-unused-import", "-Ywarn-value-discard"))
-  },
-  scalacOptions in (Test, console) := (scalacOptions in (Compile, console)).value
-)
+lazy val warnUnusedImport = Seq(scalacOptions in (Compile, console) ~= {
+  _.filterNot(Set("-Ywarn-unused-import", "-Ywarn-value-discard"))
+}, scalacOptions in (Test, console) := (scalacOptions in (Compile, console)).value)
 
-lazy val noPublishSettings = Seq(
-  publish := (),
-  publishLocal := (),
-  publishArtifact := false
-)
+lazy val noPublishSettings = Seq(publish := (), publishLocal := (), publishArtifact := false)
 
 lazy val publishSettings = Seq(
   releaseCommitMessage := s"Set version to ${if (releaseUseGlobalVersion.value) (version in ThisBuild).value
@@ -176,10 +161,7 @@ lazy val publishSettings = Seq(
   },
   autoAPIMappings := true,
   scmInfo := Some(
-    ScmInfo(
-      url("https://github.com/notxcain/aecor"),
-      "scm:git:git@github.com:notxcain/aecor.git"
-    )
+    ScmInfo(url("https://github.com/notxcain/aecor"), "scm:git:git@github.com:notxcain/aecor.git")
   ),
   pomExtra :=
     <developers>

core/src/main/resources/reference.conf

@@ -8,21 +8,10 @@ cassandra-query-journal {
 }
 
 aecor {
-  aggregate {
+  akka-runtime {
     number-of-shards = 30
     ask-timeout = 60s
-
-    default-idle-timeout = 120s
-
-    idle-timeout {
-      # per entity name idle timeout here
-    }
-
-    default-snapshot-after = off
-
-    snapshot-after {
-
-    }
+    idle-timeout = 60s
   }
 }
 
@@ -38,6 +27,15 @@ akka {
       "java.lang.String" = java
       "akka.dispatch.sysmsg.DeathWatchNotification" = java
     }
+    serialization-identifiers {
+      "aecor.aggregate.serialization.PersistentReprSerializer" = 100
+    }
+    serializers {
+      persistent-repr = "aecor.aggregate.serialization.PersistentReprSerializer"
+    }
+    serialization-bindings {
+      "aecor.aggregate.serialization.PersistentRepr" = persistent-repr
+    }
   }
   persistence {
     journal.plugin = "cassandra-journal"

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

@@ -0,0 +1,204 @@
+package aecor.aggregate
+
+import java.net.URLDecoder
+import java.nio.charset.StandardCharsets
+import java.time.{ Duration, Instant }
+
+import aecor.aggregate.SnapshotPolicy.{ EachNumberOfEvents, Never }
+import aecor.aggregate.serialization.PersistentDecoder.Result
+import aecor.aggregate.serialization.{ PersistentDecoder, PersistentEncoder, PersistentRepr }
+import aecor.data.{ Folded, Handler }
+import akka.NotUsed
+import akka.actor.{ ActorLogging, Props, ReceiveTimeout, Stash }
+import akka.cluster.sharding.ShardRegion
+import akka.persistence.journal.Tagged
+import akka.persistence.{ PersistentActor, RecoveryCompleted, SnapshotOffer }
+import cats.~>
+
+import scala.concurrent.duration.FiniteDuration
+import scala.util.{ Left, Right }
+
+sealed trait SnapshotPolicy[+E]
+
+object SnapshotPolicy {
+  def never[E]: SnapshotPolicy[E] = Never.asInstanceOf[SnapshotPolicy[E]]
+
+  def eachNumberOfEvents[E: PersistentEncoder: PersistentDecoder](
+    numberOfEvents: Int
+  ): SnapshotPolicy[E] = EachNumberOfEvents(numberOfEvents)
+
+  private[aggregate] case object Never extends SnapshotPolicy[Nothing]
+
+  private[aggregate] case class EachNumberOfEvents[State: PersistentEncoder: PersistentDecoder](
+    numberOfEvents: Int
+  ) extends SnapshotPolicy[State] {
+    def encode(state: State): PersistentRepr = PersistentEncoder[State].encode(state)
+    def decode(repr: PersistentRepr): Result[State] = PersistentDecoder[State].decode(repr)
+  }
+
+}
+
+sealed trait Identity
+object Identity {
+  case class Provided(value: String) extends Identity
+  case object FromPathName extends Identity
+}
+
+object AggregateActor {
+
+  def props[Command[_], State, Event: PersistentEncoder: PersistentDecoder](
+    entityName: String,
+    behavior: Command ~> Handler[State, Event, ?],
+    identity: Identity,
+    snapshotPolicy: SnapshotPolicy[State],
+    tagging: Tagging[Event],
+    idleTimeout: FiniteDuration
+  )(implicit folder: Folder[Folded, Event, State]): Props =
+    Props(new AggregateActor(entityName, behavior, identity, snapshotPolicy, tagging, idleTimeout))
+
+  case object Stop
+}
+
+/**
+  *
+  * Actor encapsulating state of event sourced entity behavior [Behavior]
+  *
+  * @param entityName entity name used as persistence prefix and as a tag for all events
+  * @param behavior entity behavior
+  * @param identity describes how to extract entity identifier
+  * @param snapshotPolicy snapshot policy to use
+  * @param idleTimeout - time with no commands after which graceful actor shutdown is initiated
+  */
+class AggregateActor[Command[_], State, Event: PersistentEncoder: PersistentDecoder] private[aecor] (
+  entityName: String,
+  behavior: Command ~> Handler[State, Event, ?],
+  identity: Identity,
+  snapshotPolicy: SnapshotPolicy[State],
+  tagger: Tagging[Event],
+  idleTimeout: FiniteDuration
+)(implicit folder: Folder[Folded, Event, State])
+    extends PersistentActor
+    with Stash
+    with ActorLogging {
+
+  final private val entityId: String = identity match {
+    case Identity.Provided(value) => value
+    case Identity.FromPathName =>
+      URLDecoder.decode(self.path.name, StandardCharsets.UTF_8.name())
+  }
+
+  final override val persistenceId: String = s"$entityName-$entityId"
+
+  private val recoveryStartTimestamp: Instant = Instant.now()
+
+  log.info("[{}] Starting...", persistenceId)
+
+  protected var state: State = folder.zero
+
+  private var eventCount = 0L
+
+  final override def receiveRecover: Receive = {
+    case repr: PersistentRepr =>
+      PersistentDecoder[Event].decode(repr) match {
+        case Left(cause) =>
+          onRecoveryFailure(cause, Some(repr))
+        case Right(event) =>
+          applyEvent(event)
+      }
+
+    case SnapshotOffer(_, snapshotRepr: PersistentRepr) =>
+      snapshotPolicy match {
+        case Never => ()
+        case e @ EachNumberOfEvents(_) =>
+          e.decode(snapshotRepr) match {
+            case Left(cause) =>
+              onRecoveryFailure(cause, Some(snapshotRepr))
+            case Right(snapshot) =>
+              state = snapshot
+          }
+      }
+
+    case RecoveryCompleted =>
+      log.info(
+        "[{}] Recovery to version [{}] completed in [{} ms]",
+        persistenceId,
+        lastSequenceNr,
+        Duration.between(recoveryStartTimestamp, Instant.now()).toMillis
+      )
+      setIdleTimeout()
+  }
+
+  final override def receiveCommand: Receive =
+    receivePassivationMessages.orElse(receiveCommandMessage)
+
+  private def receiveCommandMessage: Receive = {
+    case command =>
+      handleCommand(command.asInstanceOf[Command[_]])
+  }
+
+  private def handleCommand(command: Command[_]): Unit = {
+    val (events, reply) = behavior(command).run(state)
+    log.debug(
+      "[{}] Command [{}] produced reply [{}] and events [{}]",
+      persistenceId,
+      command,
+      reply,
+      events
+    )
+    val envelopes =
+      events.map(e => Tagged(PersistentEncoder[Event].encode(e), tagger(e)))
+
+    persistAll(envelopes)(_ => ())
+
+    deferAsync(NotUsed) { _ =>
+      events.foreach { event =>
+        applyEvent(event)
+        snapshotIfNeeded()
+      }
+      sender() ! reply
+    }
+  }
+
+  private def snapshotIfNeeded(): Unit =
+    snapshotPolicy match {
+      case e @ EachNumberOfEvents(numberOfEvents) if eventCount % numberOfEvents == 0 =>
+        saveSnapshot(e.encode(state))
+      case _ => ()
+    }
+
+  private def applyEvent(event: Event): Unit = {
+    state = folder
+      .fold(state, event)
+      .getOrElse {
+        val error = new IllegalStateException(s"Illegal state while applying [$event] to [$state]")
+        log.error(error, error.getMessage)
+        throw error
+      }
+    eventCount += 1
+    if (recoveryFinished)
+      log.debug("[{}] State [{}]", persistenceId, state)
+  }
+
+  private def receivePassivationMessages: Receive = {
+    case ReceiveTimeout =>
+      if (shouldPassivate) {
+        passivate()
+      } else {
+        setIdleTimeout()
+      }
+    case AggregateActor.Stop =>
+      context.stop(self)
+  }
+
+  protected def shouldPassivate: Boolean = true
+
+  private def passivate(): Unit = {
+    log.debug("[{}] Passivating...", persistenceId)
+    context.parent ! ShardRegion.Passivate(AggregateActor.Stop)
+  }
+
+  private def setIdleTimeout(): Unit = {
+    log.debug("[{}] Setting idle timeout to [{}]", persistenceId, idleTimeout)
+    context.setReceiveTimeout(idleTimeout)
+  }
+}