Estou tentando implementar o lado de leitura na minha arquitetura ES-CQRS. Digamos que eu tenha um ator persistente como este:

object UserWrite {

  sealed trait UserEvent
  sealed trait State
  case object Uninitialized extends State
  case class User(username: String, password: String) extends State
  case class AddUser(user: User)
  case class UserAdded(user: User) extends UserEvent
  case class UserEvents(userEvents: Source[(Long, UserEvent), NotUsed])
  case class UsersStream(fromSeqNo: Long)
  case object GetCurrentUser

  def props = Props(new UserWrite)
}

class UserWrite extends PersistentActor {

  import UserWrite._

  private var currentUser: State = Uninitialized

  override def persistenceId: String = "user-write"

  override def receiveRecover: Receive = {
    case UserAdded(user) => currentUser = user
  }

  override def receiveCommand: Receive = {
    case AddUser(user: User) => persist(UserAdded(user)) {
      case UserAdded(`user`) => currentUser = user
    }
    case UsersStream(fromSeqNo: Long) => publishUserEvents(fromSeqNo)
    case GetCurrentUser => sender() ! currentUser
  }

  def publishUserEvents(fromSeqNo: Long) = {
    val readJournal = PersistenceQuery(context.system).readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)
    val userEvents = readJournal
      .eventsByPersistenceId("user-write", fromSeqNo, Long.MaxValue)
      .map { case EventEnvelope(_, _, seqNo, event: UserEvent) => seqNo -> event }
    sender() ! UserEvents(userEvents)
  }
}

Pelo que entendi, sempre que um evento persistir, podemos publicá-lo viaAkka Persistence Query. Agora, não tenho certeza de qual seria a maneira correta de se inscrever nesses eventos para que eu possa persistir no meu banco de dados do lado de leitura? Uma das idéias é enviar inicialmente umUsersStream mensagem do meu ator do lado de leitura paraUserWrite ator e "afundar" eventos nesse ator lido.

EDITAR

Seguindo a sugestão do @cmbaxter, implementei o lado de leitura desta maneira:

object UserRead {

  case object GetUsers
  case class GetUserByUsername(username: String)
  case class LastProcessedEventOffset(seqNo: Long)
  case object StreamCompleted

  def props = Props(new UserRead)
}

class UserRead extends PersistentActor {
  import UserRead._

  var inMemoryUsers = Set.empty[User]
  var offset        = 0L

  override val persistenceId: String = "user-read"

  override def receiveRecover: Receive = {
    // Recovery from snapshot will always give us last sequence number
    case SnapshotOffer(_, LastProcessedEventOffset(seqNo)) => offset = seqNo
    case RecoveryCompleted                                 => recoveryCompleted()
  }

  // After recovery is being completed, events will be projected to UserRead actor
  def recoveryCompleted(): Unit = {
    implicit val materializer = ActorMaterializer()
    PersistenceQuery(context.system)
      .readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)
      .eventsByPersistenceId("user-write", offset + 1, Long.MaxValue)
      .map {
        case EventEnvelope(_, _, seqNo, event: UserEvent) => seqNo -> event
      }
      .runWith(Sink.actorRef(self, StreamCompleted))
  }

  override def receiveCommand: Receive = {
    case GetUsers                    => sender() ! inMemoryUsers
    case GetUserByUsername(username) => sender() ! inMemoryUsers.find(_.username == username)
    // Match projected event and update offset
    case (seqNo: Long, UserAdded(user)) =>
      saveSnapshot(LastProcessedEventOffset(seqNo))
      inMemoryUsers += user
  }
}

Existem alguns problemas como: O fluxo de eventos parece lento. I.e.UserRead O ator pode responder com um conjunto de usuários antes que o novo usuário adicionado seja salvo.

EDIT 2

Aumentei o intervalo de atualização do diário de consulta cassandra, que resolvia menos problemas com o fluxo de eventos lento. Parece que o diário de eventos do Cassandra é por padrão, sendo pesquisado a cada 3 segundos. No meuapplication.conf Eu adicionei:

cassandra-query-journal {
  refresh-interval = 20ms
}

EDIT 3

Na verdade, não diminua o intervalo de atualização. Isso aumentará o uso de memória, mas isso não é perigoso, nem um ponto. No conceito geral do CQRS é que o lado de gravação e leitura é assíncrono. Portanto, após a gravação, os dados nunca estarão disponíveis imediatamente para leitura. Lidando com a interface do usuário? Acabei de abrir o fluxo e enviar dados por eventos enviados pelo servidor depois que o lado de leitura os reconheceu.