多路复用 akka 流源到两个副本

Question

我正在尝试将流媒体源缓存到磁盘，同时我还将它作为HttpResponse 发送出去，即我有一个Source[ByteString,_] 我想交给HttpEntity，但我也想将相同的数据运行到 FileIO.toPath 接收器中。

                       |-> FileIO.toPath
Source[ByteString,_] ->|
                       |-> HttpEntity(contentType, Source[ByteString,_]

似乎Broadcast 是我应该用于扇出的，但根据描述，它写入两个接收器，而FileIO.toPath 是一个接收器，HttpEntity 需要一个Source。

还有Source.fromGraph，它看起来会从 GraphStage 创建一个源，例如 Broadcast 阶段，但我不太清楚如何在其中获得 FileIO 接收器。

Answer 1

你可以使用alsoTo:

val originalSource: Source[ByteString, _] = ???
val cachedSource: Source[ByteString, _] = originalSource.alsoTo(FileIO.toPath(/*...*/))
val entity = HttpEntity(contentType, cachedSource)