使用更复杂的图形时,在使发布者和订阅者脱离流程时遇到了问题。我的目标是提供发布者和订阅者的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()
有关FlowGraph.closed重载的更多信息,请参见Scaladocs。
(Keep.both
是功能的缩写(a, b) => (a, b)
)
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句