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

Question

我想在由许多操作组成的 akka 流中创建一个接收器。例如，映射，过滤，折叠然后下沉。目前我能做的最好的事情就是以下几点。我不喜欢它，因为我必须指定广播，即使我只让一个值通过。有谁知道更好的方法吗？

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)
})

}

Answer 1

akka-stream要记住的一个关键点是，任何数量的Flow值加上一个Sink值仍然是接收器。

演示此属性的几个示例：

val intSink : Sink[Int, _] = Sink.head[Int]
val anotherSink : Sink[Int, _] = 
  Flow[Int].filter(_ > 0)
           .to(intSink)
val oneMoreSink : Sink[Int, _] = 
  Flow[Int].filter(_ > 0)
           .map(_ + 4)
           .to(intSink)

因此，您可以将map和filter实现为流。 您询问的fold可以通过 Sink.fold 来实现。