Scala akka 事件溯源如何将消息返回到根目录？答案

【问题标题】：Scala akka event sourcing how to get a message back to root?Scala akka 事件溯源如何将消息返回到根目录？
【发布时间】：2022-08-18 17:39:24
【问题描述】：

我现在在努力阅读我的演员的状态，所以在这种情况下，我只想从我的 State 类中获取历史参数 - 例如在调用端点时打印它。

我已经成功地做到了？之前的操作员，但我从未尝试过事件溯源。

到目前为止，我的代码是这样的：

object MyPersistentBehavior {
  sealed trait Command
  final case class Add(data: String) extends Command
  case object Clear extends Command

  sealed trait Event
  final case class Added(data: String) extends Event
  case object Cleared extends Event

  final case class State(history: List[String] = Nil)

  val commandHandler: (State, Command) => Effect[Event, State] = { (state, command) =>
    command match {
      case Add(data) => Effect.persist(Added(data))
      case Clear     => Effect.persist(Cleared)
    }
  }

  val eventHandler: (State, Event) => State = { (state, event) =>
    event match {
      case Added(data) => state.copy((data :: state.history).take(5))
      case Cleared     => State(Nil)
    }
  }

  def apply(id: String): Behavior[Command] =
    EventSourcedBehavior[Command, Event, State](
      persistenceId = PersistenceId.ofUniqueId(id),
      emptyState = State(Nil),
      commandHandler = commandHandler,
      eventHandler = eventHandler)
}

在我的主要方法中，我想打印状态：

val personActor: ActorSystem[MyPersistentBehavior.Command] = ActorSystem(MyPersistentBehavior(\"IDDD\"), \"AHA\")
//personActor ? GetState <- something like this

谢谢！！

标签： scala state akka event-sourcing

【解决方案1】：

我没有在akka中使用过事件采购，但是快速查看了文档，我认为这可能会有所帮助：

case class GetState(replyTo: ActorRef[StatusReply[AddPostDone]]) extends Command

// and in the match for commands:
...
    case GetState(replyTo) =>
      replyTo ! StatusReply.Success(state)

      // or if replyTo was of type ActorRef[State] => 
      replyTo ! state

还有这个Effect 在akka 的事件溯源文档中看起来很有趣：

def onCommand(subscriber: ActorRef[State], state: State, command: Command): Effect[Event, State] = {
  command match {
    case Add(data) =>
      Effect.persist(Added(data)).thenRun(newState => subscriber ! newState)
    case Clear =>
      Effect.persist(Cleared).thenRun((newState: State) => subscriber ! newState).thenStop()
  }
}

如您所见，它非常易于使用。也可以在每次效果后回复最新状态：

case class AddCommand(number: Int, replyTo: ActorRef[State]) extends Command

// in the command handler
    case add: AddCommand(num, replyTo) => 
      // your logic here
      Event.persist(Added(num)) thenRun { newState => 
        replyTo ! newState
      }

文档中还有很多其他选项，所以我强烈建议您看一下：https://doc.akka.io/docs/akka/current/typed/persistence.html

【讨论】：

如何为 MyPersistentBehavior 创建 ActorSystem 实例？我好辛苦啊！
@matkenis 看看这个：akka - getting started guide

【解决方案2】：

好的，如果您需要完整的示例，请查看我的blog。

您将在博客和 GitHub 存储库中看到，在那里您将看到一个通用的“onReport”命令，所有参与者都可以向调用者报告他们的状态。

    def commandHandler(context: ActorContext[CreditSMEvent], cmd: CreditSMEvent, state: State): ReplyEffect[PersistEvent, State] =
      cmd match {
        case onReport(useCaseKey, replyTo) =>
            Effect.reply(replyTo)( ReportResponse(state, java.util.Collections.unmodifiableMap(state.controlObject)))

        case _ =>
            commandHandlerInternal(context, cmd, state)
}

您可以在博客中的 point 找到它。

我希望这会对你有所帮助。

【讨论】：