如何从操作列表中创建接收器

我想在阿克卡河上建造一个水槽,它由许多操作组成。 例如地图,过滤,折叠,然后下沉。 目前我能做的最好的事情是: 我不喜欢它,因为我必须指定广播,即使我只允许一个值通过。 有人知道更好的方法吗?

def kafkaSink(): Sink[PartialBatchProcessedResult, NotUsed] = {
    Sink.fromGraph(GraphDSL.create() { implicit b =>
    import GraphDSL.Implicits._
    val broadcast = b.add(Broadcast[PartialBatchProcessedResult](1))
    broadcast.out(0)
    .fold(new BatchPublishingResponseCollator()) { (c, e) => c.consume(e) }
    .map(_.build())
    .map(a =>
      FunctionalTesterResults(sampleProjectorConfig, 0, a)) ~> Sink.foreach(new KafkaTestResultsReporter().report)
  SinkShape(broadcast.in)
})

}