Forked source from eventuate sample

Author: Kunal Kanojia

.gitignore

@@ -0,0 +1,9 @@
+logs
+target
+/.idea
+/.idea_modules
+/.classpath
+/.project
+/.settings
+/RUNNING_PID
+/.sbtserver

.travis.yml

@@ -0,0 +1,6 @@
+language: scala
+jdk: oraclejdk8
+scala:
+   - 2.11.8
+   
+script: sbt ++$TRAVIS_SCALA_VERSION clean test

README.md

@@ -0,0 +1,33 @@
+import javax.inject._
+import play.api._
+import play.api.http.HttpFilters
+import play.api.mvc._
+
+import filters.ExampleFilter
+
+/**
+ * This class configures filters that run on every request. This
+ * class is queried by Play to get a list of filters.
+ *
+ * Play will automatically use filters from any class called
+ * `Filters` that is placed the root package. You can load filters
+ * from a different class by adding a `play.http.filters` setting to
+ * the `application.conf` configuration file.
+ *
+ * @param env Basic environment settings for the current application.
+ * @param exampleFilter A demonstration filter that adds a header to
+ * each response.
+ */
+@Singleton
+class Filters @Inject() (
+  env: Environment,
+  exampleFilter: ExampleFilter) extends HttpFilters {
+
+  override val filters = {
+    // Use the example filter if we're running development mode. If
+    // we're running in production or test mode then don't use any
+    // filters at all.
+    if (env.mode == Mode.Dev) Seq(exampleFilter) else Seq.empty
+  }
+
+}

app/Filters.scala

@@ -0,0 +1,33 @@
+import java.time.Clock
+
+import com.google.inject.AbstractModule
+import com.kkanojia.example.actors.{WSUserActor, WSUserParentActor}
+import com.kkanojia.example.modules.ActorSystemInitializer
+
+import play.api.libs.concurrent.AkkaGuiceSupport
+
+import services.ApplicationTimer
+
+/**
+ * This class is a Guice module that tells Guice how to bind several
+ * different types. This Guice module is created when the Play
+ * application starts.
+ *
+ * Play will automatically use any class called `Module` that is in
+ * the root package. You can create modules in other locations by
+ * adding `play.modules.enabled` settings to the `application.conf`
+ * configuration file.
+ */
+class Module extends AbstractModule with AkkaGuiceSupport {
+
+  override def configure() = {
+    bind(classOf[ActorSystemInitializer]).asEagerSingleton()
+
+    bind(classOf[Clock]).toInstance(Clock.systemDefaultZone)
+    bind(classOf[ApplicationTimer]).asEagerSingleton()
+
+    bindActor[WSUserParentActor]("userParentActor")
+    bindActorFactory[WSUserActor, WSUserActor.Factory]
+  }
+
+}

app/Module.scala

@@ -0,0 +1,3 @@
+package com.kkanojia.example.actors
+
+trait TaggedEvent

app/com/kkanojia/example/actors/TaggedEvent.scala

@@ -0,0 +1,18 @@
+package com.kkanojia.example.actors
+
+import akka.persistence.journal.{Tagged, WriteEventAdapter}
+import com.kkanojia.example.actors.TradeActor.{TradeCreated, TradeUpdated}
+import com.kkanojia.example.actors.UserActor.UserCreated
+
+class TaggingEventAdapter extends WriteEventAdapter {
+
+  override def toJournal(event: Any): Any = event match {
+    case e: UserCreated => Tagged(event, Set("user-events"))
+    case e: TradeCreated => Tagged(event, Set("trade-events"))
+    case e: TradeUpdated => Tagged(event, Set("trade-events"))
+    case _ => event
+  }
+
+  override def manifest(event: Any): String = ""
+
+}

app/com/kkanojia/example/actors/TaggingEventAdapter.scala

@@ -0,0 +1,60 @@
+package com.kkanojia.example.actors
+
+import akka.persistence.PersistentActor
+import akka.persistence.journal.leveldb.SharedLeveldbJournal
+import com.kkanojia.example.actors.TradeActor._
+import com.kkanojia.example.models.Trade
+
+object TradeActor {
+
+  //Commands
+  case class CreateTrade(trade: Trade)
+
+  case class UpdateTrade(trade: Trade)
+
+  //Replies
+  case class CreateTradeSuccess(trade: Trade)
+
+  case class CreateTradeFailure(cause: Throwable)
+
+  case class UpdateTradeSuccess(trade: Trade)
+
+  case class UpdateTradeFailure(cause: Throwable)
+
+  //Events
+  case class TradeCreated(trade: Trade) extends TaggedEvent
+
+  case class TradeUpdated(trade: Trade) extends TaggedEvent
+
+}
+
+class TradeActor(val id: String
+) extends PersistentActor {
+
+  override def persistenceId = id
+
+  private var tradeOpt: Option[Trade] = None
+
+  SharedLeveldbJournal.setStore(context.self, context.system)
+
+  override def receiveCommand: Receive = {
+
+    case CreateTrade(trade) =>
+      persist(TradeCreated(trade)) {
+       evt => sender() ! CreateTradeSuccess(trade)
+      }
+
+    case UpdateTrade(trade) =>
+      persist(TradeUpdated(trade)) {
+        evt => sender() ! UpdateTradeSuccess(trade)
+      }
+  }
+
+  override def receiveRecover: Receive = {
+    case TradeCreated(trade) =>
+      tradeOpt = Some(trade)
+
+    case TradeUpdated(trade) =>
+      tradeOpt = Some(trade)
+  }
+}

app/com/kkanojia/example/actors/TradeActor.scala

@@ -0,0 +1,49 @@
+package com.kkanojia.example.actors
+
+import scala.collection.immutable.HashSet
+import scala.collection.mutable
+
+import akka.actor.{Actor, ActorRef, Stash}
+import com.kkanojia.example.actors.TradeActor.{TradeCreated, TradeUpdated}
+import com.kkanojia.example.actors.TradeAggregateViewActor.{UnWatchTrades, WatchTrades}
+import com.kkanojia.example.models.Trade
+
+object TradeAggregateViewActor {
+
+  val ID = "464788cb-58aa-4dc6-8dce-703a456c238a"
+  val NAME = "trade_view_aggregate"
+
+  case object WatchTrades
+
+  case object UnWatchTrades
+
+}
+
+class TradeAggregateViewActor(val id: String
+) extends Actor {
+
+  protected[this] var watchers = HashSet.empty[ActorRef]
+
+  private val trades = mutable.Map[String, Trade]()
+
+  override def receive: Receive = {
+
+    case WatchTrades =>
+      watchers = watchers + sender
+
+    case UnWatchTrades =>
+      watchers = watchers - sender
+
+  }
+
+  def onEvent: Receive = {
+    case TradeCreated(trade) =>
+      trades(trade.id) = trade
+      watchers.foreach(_ ! TradeCreated(trade))
+
+    case TradeUpdated(trade) =>
+      trades(trade.id) = trade
+      watchers.foreach(_ ! TradeUpdated(trade))
+
+  }
+}

app/com/kkanojia/example/actors/TradeAggregateViewActor.scala

@@ -0,0 +1,75 @@
+package com.kkanojia.example.actors
+
+import scala.collection.mutable
+
+import akka.NotUsed
+import akka.actor.{Actor, ActorRef, Props}
+import akka.persistence.query.{EventEnvelope, PersistenceQuery}
+import akka.persistence.query.journal.leveldb.scaladsl.LeveldbReadJournal
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.Source
+import com.kkanojia.example.actors.UserActor.UserCreated
+import com.kkanojia.example.models.Trade
+
+object TradeManager {
+
+  //Command
+  case object RetrieveTrades
+
+  case class FindTrade(tradeId: String)
+
+  //Replies
+  case class RetrieveTradesSuccess(trades: Seq[Trade])
+
+  case class FindTradeSuccess(tradeOpt: Option[Trade])
+
+}
+
+class TradeManager(val id: String) extends Actor {
+
+  import TradeActor._
+  import TradeManager._
+
+  private val userTrades = mutable.Map[String, Trade]()
+
+  val queries = PersistenceQuery(context.system).readJournalFor[LeveldbReadJournal](
+    LeveldbReadJournal.Identifier)
+
+  val src: Source[EventEnvelope, NotUsed] = queries.eventsByTag("trade-events", 0L)
+
+  implicit val mat = ActorMaterializer()
+  src.runForeach { env =>
+    env.event match {
+      case TradeCreated(trade) =>
+        userTrades(trade.id) = trade
+
+      case TradeUpdated(trade) =>
+        userTrades(trade.id) = trade
+
+      case _ => println(s"Unknown event $env")
+    }}
+
+  override def receive: Receive = {
+
+    case CreateTrade(trade) =>
+      getTradeActor(trade.id) forward CreateTrade(trade)
+
+    case RetrieveTrades =>
+      sender() ! RetrieveTradesSuccess(userTrades.values.toSeq)
+
+    case FindTrade(tradeId: String) =>
+      sender() ! FindTradeSuccess(userTrades.get(tradeId))
+
+    case UpdateTrade(trade) =>
+      getTradeActor(trade.id) forward UpdateTrade(trade)
+
+  }
+
+  private def getTradeActor(tradeId: String): ActorRef = {
+    val name = s"trade_$tradeId"
+    context.child(name) match {
+      case Some(actorRef) => actorRef
+      case None => context.actorOf(Props(new TradeActor(tradeId)), name)
+    }
+  }
+}