从 Akka 中的轮询方法创建 Source

Question

如果有对新元素的请求，我有一个方法可以使用 poll 方法从不同来源获取元素。我怎样才能从这个方法创建源，以便它只从轮询方法中请求一个新元素？

Answer 1

akka.io 博客文章中解释了与轮询 API 的集成：Writing Akka Streams Connectors for existing APIs 在“基于轮询的 API”部分中。

在它的核心，您需要扩展 TimerGraphStageLogic，并执行以下操作：

private void schedulePoll() {
  scheduleOnce("poll", pollInterval);
}

@Override
public void onTimer(Object timerKey) {
  if (!isClosed(out)) {
    doPoll();
    if (!buffer.isEmpty()) {
      pushHead();
    } else {
      schedulePoll();
    }
  }
}

安排轮询。

或者，您可以坚持在unfoldAsyncResource 中实现所有回调。

这个阶段的完整实现可以在here找到（Java实现）。