Akka Streams：在GraphStage源代码中处理未来

Question

我试图构建一个Akka流源，它通过发出futureAPI调用来接收数据（API的本质是滚动，它增量地获取结果）。为了构建这样的源码，我使用了GraphStage。

我修改了NumberSource示例，它只是一次推送一个int。我所做的唯一更改是将Int替换为getvalue():future[Int]（以模拟API调用）：

class NumbersSource extends GraphStage[SourceShape[Int]] {
  val out: Outlet[Int] = Outlet("NumbersSource")
  override val shape: SourceShape[Int] = SourceShape(out)

  // simple example of future API call
  private def getvalue(): Future[Int] = Future.successful(Random.nextInt())


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

      setHandler(out, new OutHandler {
        override def onPull(): Unit = {
            // Future API call
            getvalue().onComplete{
              case Success(value) =>
                println("Pushing value received..") // this is currently being printed just once
                push(out, counter)
              case Failure(exception) =>
            }
          }
        }
      })
    }
}

// Using the Source and Running the stream

  val sourceGraph: Graph[SourceShape[Int], NotUsed] = new NumbersSource
  val mySource: Source[Int, NotUsed] = Source.fromGraph(sourceGraph)

  val done: Future[Done] = mySource.runForeach{
    num => println(s"Received: $num") // This is currently not printed
  }
  done.onComplete(_ => system.terminate())

上面的代码不起作用。sethandler中的println语句只执行一次，不向下游推送任何内容。

这类未来来电应如何处理？谢了。

class NumbersSource(futureNum: Future[Int]) extends GraphStage[SourceShape[Int]] {
  val out: Outlet[Int] = Outlet("NumbersSource")
  override val shape: SourceShape[Int] = SourceShape(out)

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

      override def preStart(): Unit = {
        val callback = getAsyncCallback[Int] { (_) =>
          completeStage()
        }
        futureNum.foreach(callback.invoke)
      }

      setHandler(out, new OutHandler {
        override def onPull(): Unit = {
          val value: Int = ??? // How to get this value ??
          push(out, value)
        }
      })
    }
}

// Using the Source and Running the Stream

def random(): Future[Int] = Future.successful(Random.nextInt())

  val sourceGraph: Graph[SourceShape[Int], NotUsed] = new NumbersSource(random())
val mySource: Source[Int, NotUsed] = Source.fromGraph(sourceGraph)

  val done: Future[Done] = mySource.runForeach{
    num => println(s"Received: $num") // This is currently not printed
  }
  done.onComplete(_ => system.terminate())

val value = grab(in) // where in is Inlet of a Flow

Answer 1

我不确定我的理解是否正确，但是如果您试图实现一个无限的源代码，而不是在Futures中计算的元素，那么实际上没有必要使用您自己的GraphStage来实现这个源代码。您可以简单地按以下方式进行：

Source.repeat(())
    .mapAsync(parallelism) { _ => Future.successful(Random.nextInt()) }

source.repeat(())只是一些任意值的无限源（在本例中为unit类型，但是您可以将()更改为您想要的任何值，因为这里忽略了它）。然后使用MapAsync将异步计算集成到流中。