如何将 akka.stream.javadsl.Source 转换为 java.util.concurrent.Flow

Question

我正在试验反应式流的Test Compatability Kit (tck)，并且自己测试了一些发布者，我想测试一个akka Source。但是为了做到这一点，我需要将源(或源+处理器)转换为java.util.concurrent.Flow.Publisher。

@Override
public Flow.Publisher<Integer> createFlowPublisher(long elements) {
return new FlowPublisher((int) elements);  // <-- how to test an akka.Source?
}
class FlowPublisher implements java.util.concurrent.Flow.Publisher {
...

而且我在任何地方都找不到如何做到这一点。

是否有一些关于此的文档，或者有人知道答案？

Answer 1

您可以将 akka-streamsSource转换为org.reactivestreams.Publisher(这与根据反应流java.util.concurrent.Flow相同)：

implicit val sys: ActorSystem = ActorSystem()
implicit val mat: ActorMaterializer = ActorMaterializer()
val source = Source.single(1) // some random source
val publisher: Publisher[Int] = source.runWith(Sink.asPublisher(false))

您可以在官方 akka-streams 文档中找到详细信息 - 与反应式流集成

---

需要注意的是，Akka-Streams有文档和一些用于测试其组件的工具包 - 测试流。 我想说的是，最好使用 AKKA-Streams 测试套件测试 AKKA-Streams 组件，因为您无需对org.reactivestreams.*实体进行额外的转换，并且测试代码将更加干净、简单和可靠