scala - 如何从广播的Akka流中获取订阅者和发布者？

使用更复杂的图形时，在使发布者和订阅者脱离流程时遇到了问题。我的目标是提供发布者和订阅者的API，并在内部运行Akka流。这是我的第一次尝试，效果很好。

val subscriberSource = Source.subscriber[Boolean]
val someFunctionSink = Sink.foreach(Console.println)

val flow = subscriberSource.to(someFunctionSink)

//create Reactive Streams Subscriber
val subscriber: Subscriber[Boolean] = flow.run()

//prints true
Source.single(true).to(Sink(subscriber)).run()

但是，对于更复杂的广播图，我不确定如何获取Subscriber和Publisher对象？我需要局部图吗？

val subscriberSource = Source.subscriber[Boolean]
val someFunctionSink = Sink.foreach(Console.println)
val publisherSink = Sink.publisher[Boolean]

FlowGraph.closed() { implicit builder =>
  import FlowGraph.Implicits._

  val broadcast = builder.add(Broadcast[Boolean](2))

  subscriberSource ~> broadcast.in
  broadcast.out(0) ~> someFunctionSink
  broadcast.out(1) ~> publisherSink
}.run()

val subscriber: Subscriber[Boolean] = ???
val publisher: Publisher[Boolean] = ???

最佳答案

当您调用RunnableGraph.run()时，将运行流，结果是该运行的“物化值”。

在您的简单示例中，Source.subscriber[Boolean]的物化值为Subscriber[Boolean]。在您的复杂示例中，您想要将图形的几个组成部分的物化值合并为一个元组(Subscriber[Boolean], Publisher[Boolean])的物化值。

您可以通过将您对其物化值感兴趣的组件传递给FlowGraph.closed()，然后指定一个函数来组合物化值来实现:

import akka.stream.scaladsl._
import org.reactivestreams._

val subscriberSource = Source.subscriber[Boolean]
val someFunctionSink = Sink.foreach(Console.println)
val publisherSink = Sink.publisher[Boolean]

val graph =
  FlowGraph.closed(subscriberSource, publisherSink)(Keep.both) { implicit builder ⇒
    (in, out) ⇒
      import FlowGraph.Implicits._

      val broadcast = builder.add(Broadcast[Boolean](2))

      in ~> broadcast.in
      broadcast.out(0) ~> someFunctionSink
      broadcast.out(1) ~> out
  }
val (subscriber: Subscriber[Boolean], publisher: Publisher[Boolean]) = graph.run()

参见Scaladocs for more information about the overloads of FlowGraph.closed。

(Keep.both是函数(a, b) => (a, b)的缩写)

关于scala - 如何从广播的Akka流中获取订阅者和发布者？，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/31195254/