EventSourcedBehavior: Initialize "actor" with external data

Akka Persistence / Event Sourcing
1

I am trying to set up an EventSourcedBehavior whose State is a combination of data retrieved from a DB and events consumed from Kafka. How/where can I inject the former? I can imagine two options:

  1. When the behavior is initialized, send a Load Command to it from the outside. I don’t see how this is possible a) because I have no reference to the “actorRef” anywhere that I can see (following the ShoppingCart approach) and b) the Behavior is initialized in response to the first Kafka message, so presumably that message is going to be first in the mailbox

  2. In the CommandHandler, check for whether we are fully initialized, and if not, stash the current command, initialize the data from the DB, and then unstash. I am trying to implement this approach, but the limited Effect options don’t seem to support that. What I have is

  private def handleCommand(
      context: ActorContext[Command],
      state: State,
      command: Command
  )(implicit ec: ExecutionContext): Effect[Event, State] = {
    if (state.isInitialized) {
      initializedCommandHandler(context, sessionId, state, command)
    } else {
      uninitializedCommandHandler(state)
    }
  }

  private def initializedCommandHandler(
      context: ActorContext[Command],
      sessionId: String,
      state: State,
      command: Command
  ): Effect[Event, State] = {
    command match {
      case Update(_)   =>
      case ShutDown(_) =>
    }
  }

  private def uninitializedCommandHandler(
      state: State
  )(implicit ec: ExecutionContext): Effect[Event, State] = {
    Effect.stash()
    val targets = loadFromDb(state)
    Effect.persist(Initialized(targets)).thenUnstashAll()
  }

I was looking for some way to do something like this.

Effect.stash
.thenRun(loadFromDb()) // does not compile - no such option, contrary to docs on stash method
.andThen(data => Effect.persist(Initialized(data)).thenUnstashAll()

What would be the correct way to do this?

Thanks!

2

An effect can not return an effect as a side effect, it would have to send a command to itself on completion of the async task.

One way to implement this that could be a bit cleaner is to keep the stashing when state not initialized, but instead of trigger load on command react on recovery completing see docs, trigger async load of data if not initialized, once async load load of the external data completes, turn it into an “Initialize”-command and send to self, let the command handler for it persist an event with the initial state, after that has been applied the state is initialized, so can thenUnstashAll() as a side effect.

3

This is a nice idea. I have done something very similar with a Hazelcast system. Thanks!