Akka-stream 从 Cassandra 中过滤掉特定事件答案

【问题标题】：Akka-stream filter out particular event from CassandraAkka-stream 从 Cassandra 中过滤掉特定事件
【发布时间】：2017-05-04 16:25:14
【问题描述】：

我目前有一个由 Akka 和 Cassandra 支持的事件源服务。这是一个名为 AuctionService 的投标系统，有时我需要检索最后一个名为 BiddenOnLot 的投标事件。为此我使用akka-persistence-query。

这是我当前的代码：

        // obtain read journal by plugin id
        val readJournal = PersistenceQuery(context.system).readJournalFor[CassandraReadJournal](
          "cassandra-query-journal")

        // issue query to journal
        val source: Source[EventEnvelope, NotUsed] = readJournal.eventsByPersistenceId(self.path.name.toString, 0, Long.MaxValue)

        // materialize stream, consuming events
        implicit val mat = ActorMaterializer()

        source.runForeach(envelope ⇒ {
          if (envelope.event.isInstanceOf[BiddenOnLot]) {
            val biddenOnLot = envelope.event.asInstanceOf[BiddenOnLot]
            if (biddenOnLot.paddleId == paddleId) {
               // TODO: get last bid event by paddle id
            }
          }
        })

到目前为止，我遍历了所有事件并可以确定其类型。但我真的很难隔离最后一次出价事件并能够异步使用它。有什么想法吗？

【问题讨论】：

标签： scala akka akka-stream akka-persistence akka-persistent-fsm

【解决方案1】：

您可以通过使用模式匹配来解构发出的EventEnvelope 并将其映射到您想要的事件：

source.map {
    case EventEnvelope(_, _, _, bidenOnLot: BidenOnLot) if bidenOnLot.paddleId == paddleId => bidenOnLot
}.runForeach(println)

【讨论】：