我想用这样的逻辑实现 Diff 转换流：
我有一些 Source[JsValue] 和一些带有 id->JsValue 数据的数据库。我想在Source和db中从JsValue获取Diff，然后进一步拉取生成的Diff并将JsValue从源存储到db。
我认为 Akka-persistence 作为存储实现，但我只需要当前状态，所以它可以是任何键值数据库。
由于我是 akka-stream 的新手，我不明白实现这个想法的最佳方法是什么。
【问题讨论】：
标签： scala akka akka-stream
假设您的JsValue 对象有一个“id”字段，您可以编写一个查询Flow，它将接收原始JsValue 并生成原始和数据库版本的元组：
def dbQuery(id : String) : JsValue = ???
val queryFlow : Flow[JsValue, (JsValue,JsValue), _] =
Flow[JsValue] map { originalJs =>
originalJs -> dbQuery((originalJs \ "id").as[String])
}
这些元组可以传递到一个不同的Flow：
def diffJs(original : JsValue, dbVersion : JsValue) : JsValue = ???
val diffFlow : Flow[(JsValue, JsValue), JsValue, _] =
Flow[(JsValue, JsValue)] map diffJs.tupled
你提到的最后一部分是一个数据库，它将保持差异，this can represented 和Sink：
val dbSink : Sink[JsValue, _] = ???
然后可以根据您的值源组合所有这些组件以形成流：
val jsSource : Source[JsValue, _] = ???
jsSource via queryFlow via diffFlow runWith dbSink
有关异步数据库查询的示例，请参阅 this example 演示 mapAsync。
【讨论】：
dbQuery: Future[JsValue] 怎么样？我不能使用简单的map。