Akka-Effect

This project aims to build a bridge between akka and pure functional code based on cats-effect

Covered:

• Actors
• Persistence

Building blocks

akka-effect-actor module

Tell.scala

Represents ActorRef.tell

trait Tell[F[_], -A] {

  def apply(a: A, sender: Option[ActorRef] = None): F[Unit]
}

Ask.scala

Represents ActorRef.ask pattern

trait Ask[F[_], -A, B] {

  def apply(msg: A, timeout: FiniteDuration, sender: Option[ActorRef]): F[B]
}

Reply.scala

Represents reply pattern: sender() ! reply

trait Reply[F[_], -A] {

  def apply(msg: A): F[Unit]
}

Receive.scala

This is what you need to implement instead of familiar new Actor { ... }

trait Receive[F[_], -A, B] {

  def apply(msg: A): F[B]

  def timeout:  F[B]
}

ActorOf.scala

Constructs Actor.scala out of receive: ActorCtx[F] => Resource[F, Receive[F, Any]]

ActorCtx.scala

Wraps ActorContext

trait ActorCtx[F[_]] {

  def self: ActorRef

  def parent: ActorRef

  def executor: ExecutionContextExecutor

  def setReceiveTimeout(timeout: Duration): F[Unit]

  def child(name: String): F[Option[ActorRef]]

  def children: F[List[ActorRef]]

  def actorRefFactory: ActorRefFactory

  def watch[A](actorRef: ActorRef, msg: A): F[Unit]

  def unwatch(actorRef: ActorRef): F[Unit]

  def stop: F[Unit]
}

akka-effect-persistence module

PersistentActorOf.scala

Constructs PersistentActor.scala out of eventSourcedOf: ActorCtx[F] => F[EventSourced[F, S, E, C]]

EventSourced.scala

Describes a lifecycle of entity with regard to event sourcing, phases are: Started, Recovering, Receiving and Termination

trait EventSourced[F[_], S, E, C] {

  def eventSourcedId: EventSourcedId

  def recovery: Recovery

  def pluginIds: PluginIds

  def start: Resource[F, RecoveryStarted[F, S, E, C]]
}

RecoveryStarted.scala

Describes start of recovery phase

trait RecoveryStarted[F[_], S, E, C] {

  def apply(
    seqNr: SeqNr,
    snapshotOffer: Option[SnapshotOffer[S]]
  ): Resource[F, Recovering[F, S, E, C]]
}

Recovering.scala

Describes recovery phase

trait Recovering[F[_], S, E, C] {

  def replay: Resource[F, Replay[F, E]]

  def completed(
    seqNr: SeqNr,
    journaller: Journaller[F, E],
    snapshotter: Snapshotter[F, S]
  ): Resource[F, Receive[F, C]]
}

Replay.scala

Used during recovery to replay events

trait Replay[F[_], A] {

  def apply(seqNr: SeqNr, event: A): F[Unit]
}

Journaller.scala

Describes communication with underlying journal

trait Journaller[F[_], -A] {

  def append: Append[F, A]

  def deleteTo: DeleteEventsTo[F]
}

Snapshotter.scala

Describes communication with underlying snapshot storage

/**
  * Describes communication with underlying snapshot storage
  *
  * @tparam A - snapshot
  */
trait Snapshotter[F[_], -A] {

  def save(seqNr: SeqNr, snapshot: A): F[F[Instant]]

  def delete(seqNr: SeqNr): F[F[Unit]]

  def delete(criteria: SnapshotSelectionCriteria): F[F[Unit]]
}

akka-effect-eventsourced module

TODO

Setup

in build.sbt

addSbtPlugin("com.evolution" % "sbt-artifactory-plugin" % "0.0.2")

libraryDependencies += "com.evolutiongaming" %% "akka-effect-actor" % "0.2.0"

libraryDependencies += "com.evolutiongaming" %% "akka-effect-persistence" % "0.2.0"