【发布时间】:2016-02-29 02:42:28
【问题描述】:
我使用图形 dsl 根据我看到的一些示例代码创建了一些流处理作业。一切运行良好,我只是无法理解符号:(更新为 2.4)
def elements: Source[Foos] = ...
def logEveryNSink = // a sink that logs
def cleaner: Flow[Foos, Bars, Unit] = ...
def boolChecker(bar: Bar)(implicit ex: ExecutionContext): Future[Boolean] = ...
val mySink = Sink.foreach[Boolean](println(_))
val lastly = Flow[Bars].mapAsync(2)(x => boolChecker(x).toMat(mySink)(Keep.right)
val materialized = RunnableGraph.fromGraph(
GraphDSL.create(lastly) { implicit builder =>
baz => {
import GraphDSL.Implicits._
val broadcast1 = builder.add(Broadcast[Foos](2))
val broadcast2 = builder.add(Broadcast[Bars](2))
elements ~> broadcast1 ~> logEveryNSink(1)
broadcast1 ~> cleaner ~> broadcast2 ~> baz
~> broadcast2 ~> logEveryNSink(1)
ClosedShape
}
}
).run()
我了解其中包含的隐式构建器,但我不确定baz 在{ implicit builder => baz => { ... 中代表什么。它只是整个形状的隐含名称吗?
【问题讨论】:
标签: akka akka-stream