Akka溪流-可暂停图形舞台（Akka 2.5.7）

Question

我想写一个GraphStage，可以通过发送另一个演员的消息来暂停/取消暂停。

下面截取的代码显示了一个简单的GraphStage，它会生成随机数。当舞台具体化时，GraphStageLogic会向主管发送一条包含StageActor的消息（在preStart（）中）。主管保留舞台的ActorRef，因此可用于控制舞台。

object RandomNumberSource {
  case object Pause
  case object UnPause
}

class RandomNumberSource(supervisor: ActorRef) extends GraphStage[SourceShape[Int]] {

  val out: Outlet[Int] = Outlet("rnd.out")

  override val shape: SourceShape[Int] = SourceShape(out)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = {
    new RandomNumberSourceLogic(shape)
  }

  private class RandomNumberSourceLogic(shape: Shape) extends GraphStageLogic(shape) with StageLogging {

    lazy val self: StageActor = getStageActor(onMessage)

    val numberGenerator: Random = Random
    var isPaused: Boolean = true

      override def preStart(): Unit = {
        supervisor ! AssignStageActor(self.ref)
      }

      setHandler(out, new OutHandler {
        override def onPull(): Unit = {
          if (!isPaused) {
            push(out, numberGenerator.nextInt())
            Thread.sleep(1000)
          }
        }
      })

      private def onMessage(x: (ActorRef, Any)): Unit =
      {
        x._2 match {
          case Pause =>
            isPaused = true
            log.info("Stream paused")
          case UnPause =>
            isPaused = false
            getHandler(out).onPull()
            log.info("Stream unpaused!")
          case _ =>
        }
      }
    }
}

这是supervisor actor的一个非常简单的实现：

object Supervisor {
  case class AssignStageActor(ref: ActorRef)
}

class Supervisor extends Actor with ActorLogging {

  var stageActor: Option[ActorRef] = None

  override def receive: Receive = {

    case AssignStageActor(ref) =>
      log.info("Stage assigned!")
      stageActor = Some(ref)
      ref ! Done

    case Pause =>
      log.info("Pause stream!")
      stageActor match {
        case Some(ref) => ref ! Pause
        case _ =>
      }

    case UnPause =>
      log.info("UnPause stream!")
      stageActor match {
        case Some(ref) => ref ! UnPause
        case _ =>
      }
  }
}

我正在使用以下应用程序运行流：

object Application extends App {

  implicit val system = ActorSystem("my-actor-system")
  implicit val materializer = ActorMaterializer()

  val supervisor = system.actorOf(Props[Supervisor], "supervisor")

  val sourceGraph: Graph[SourceShape[Int], NotUsed] = new RandomNumberSource(supervisor)
  val randomNumberSource: Source[Int, NotUsed] = Source.fromGraph(sourceGraph)

  randomNumberSource.take(100).runForeach(println)

  println("Start stream by pressing any key")

  StdIn.readLine()

  supervisor ! UnPause

  StdIn.readLine()

  supervisor ! Pause

  StdIn.readLine()

  println("=== Terminating ===")
  system.terminate()
}

当应用程序以“暂停”状态启动阶段ia并且不产生任何数字时。当我按下键时，我的舞台开始发出数字。但我的问题是，在它启动后发送到舞台的所有消息都被忽略了。我无法暂停阶段。

我感兴趣的是根据从演员那里收到的消息来改变舞台的行为，但我发现的所有示例都将演员的消息传递到流中。

有人猜到我的代码为什么不起作用了吗？或者有人知道如何构建这样一个GraphStage？

非常感谢你！

Answer 1

Akka Stream Contrib项目有一个阀阶段，它具体化了一个可以暂停和恢复流量的值。从这个类的Scaladoc中：

具体化为未来的ValveSwitch，它提供了一个方法翻转，停止或重新启动通过舞台的元素流。只要阀门关闭，它就会产生背压。

例如：

val (switchFut, seqSink) = Source(1 to 10)
  .viaMat(new Valve(SwitchMode.Close))(Keep.right)
  .toMat(Sink.seq)(Keep.both)
  .run()

switchFut是一个未来的[Valves switch]，由于开关最初是关闭的，因此阀门背压和任何东西都不会排放到下游。要打开阀门：

switchFut.onComplete {
  case Success(switch) =>  
    switch.flip(SwitchMode.Open) // Future[Boolean]
  case _ =>
    log.error("the valve failed")
}

更多示例见ValveSpec。