【发布时间】:2020-08-14 00:01:46
【问题描述】:
我有一些代码大致如下所示:其中 A 是两个地图的元组
def methodName(): Flow[A, B, NotUsed] = {
val filter = Flow[A].map(a => a._2.slice(0, 2))
val split = Flow[A._2]
.mapConcat(identity)
.map(t => {
B.random
})
.fold(B.empty)((a, b) => {
new B(a._1, a._2 ++ Seq(b._1), a._3 ++ Seq(b._2), a._4)
})
val logK = Flow[B].log("K", c => {
log.info("here")
})
filter.via(split).via(logK)
}
但是当我运行这个时,流在折叠阶段停止,我不明白为什么。我可以确认 A._2 中的集合已完全用尽,当我用不同的操作替换折叠时,流程继续进行并且没有被阻塞。据我所知,上游mapConcat正在调用completeStage。所以我不确定为什么 fold 阶段没有接到那个电话并且知道继续下一个阶段。
【问题讨论】:
标签: scala akka-stream