scala – 如何透明地将输入元素与输出元素相关联

2023年10月14日 179次阅读

我得到一个流< A,B> (这是一个奇特的流/图形东西,见
https://doc.akka.io/api/akka/current/akka/stream/scaladsl/Flow.html)来自我控制之外的一些外部代码.我需要包装该流并对每个输入元素和每个输出元素进行一些处理.我可以通过在它上面放置一个
BidiFlow来轻松实现这一点：

Flow<I, O, Unused> flow = ...; // external source
BidiFlow<I, I, O, O, Unused> bidi = BidiFlow.fromFunctions(i -> preprocess(i), o -> postprocess(o)); // do something on every input and every output
Flow<I, O, Unused> newFlow = bidi.join(flow);

所以这是扭曲：要正确地后处理输出元素o,我需要生成该输出元素的输入.由于我无法控制底层流,因此我无法重构它以返回,例如,输入和输出的元组.由于Akka的异步和并行特性,我不能做任何技巧,比如将输入存储在本地线程或静态字段或类似的东西上.

所以我的问题是：是否有一些Akka Streams魔术可以应用于某种方式获取生成输出的输入元素？

最佳答案 这是使用GraphDsl,Broadcast和Zip阶段的解决方案.

  val externalFlow: Flow[Int, String, NotUsed] = Flow[Int].map(i => i.toString + "-external")

  def zipInAndOut[I, O](flow: Flow[I, O, NotUsed]): Flow[I, (I, O), NotUsed] = {
    Flow.fromGraph(GraphDSL.create() { implicit b =>
      import GraphDSL.Implicits._
      val broadcast = b.add(Broadcast[I](2))
      val zip = b.add(Zip[I, O])
      val theFlow = b.add(flow)
      broadcast.out(0) ~> zip.in0
      broadcast.out(1) ~> theFlow ~> zip.in1
      new FlowShape(broadcast.in, zip.out)
    })
  }
  Source
    .fromIterator(() => (1 until 10).iterator)
    .via(zipInAndOut(externalFlow))
    .runWith(Sink.foreach(println))

版画

(1,1-external)
(2,2-external)
(3,3-external)
(4,4-external)
(5,5-external)
(6,6-external)
(7,7-external)
(8,8-external)
(9,9-external)
点赞