斯卡拉 – 如何从外面正确阻止阿卡流

剩下的问题是,如何将其包含在无限的随机数流中.我试过Zip.结果是没有随机数通过流,因为lazyEmpty从不发出值(doh).我尝试了Merge,但是流程从未终止,因为Merge一直持续到所有来源都已完成.

所以我写了自己的合并.它从一个输入端口转发所有值,并在任何源完成时终止.

object StopperFlow {

  private class StopperMergeShape[A](_init: Init[A] = Name("StopperFlow")) extends FanInShape[A](_init) {
    val in = newInlet[A]("in")
    val stopper = newInlet[Unit]("stopper")

    override protected def construct(init: Init[A]): FanInShape[A] = new StopperMergeShape[A](init)
  }

  private class StopperMerge[In] extends FlexiMerge[In,StopperMergeShape[In]](
    new StopperMergeShape(),Attributes.name("StopperMerge")) {
    import FlexiMerge._

    override def createMergeLogic(p: PortT) = new MergeLogic[In] {
      override def initialState =
        State[In](Read(p.in)) { (ctx,input,element) =>
          ctx.emit(element)
          SameState
        }

      override def initialCompletionHandling = eagerClose
    }
  }

  def apply[In](): Flow[In,In,Promise[Unit]] = {
    val stopperSource = Source.lazyEmpty[Unit]

    Flow(stopperSource) { implicit builder =>
      stopper =>
        val stopperMerge = builder.add(new StopperMerge[In]())

        stopper ~> stopperMerge.stopper

        (stopperMerge.in,stopperMerge.out)
    }
  }    
}

流可以插入任何流.实现后,它将返回一个Promise,在完成时终止流.这是我对它的测试.

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()

val startTime = System.currentTimeMillis()

def dumpToConsole(f: Float) = {
  val timeSinceStart = System.currentTimeMillis() - startTime
  System.out.println(s"[$timeSinceStart] - Random number: $f")
}

val randomSource = Source(() => Iterator.continually(Random.nextFloat()))
val consoleSink = Sink.foreach(dumpToConsole)
val flow = randomSource.viaMat(StopperFlow())(Keep.both).to(consoleSink)

val (_,promise) = flow.run()

Thread.sleep(1000)
val _ = promise.success(())
Thread.sleep(1000)

我希望这对其他人也有用.仍然让我感到困惑的是,为什么没有内置的方法来终止来自流外部的流.