使用Akka通过WebSocket流化分页API响应

Question

在服务器端，我正在使用一个HTTP API，它以页面形式返回结果。如中所示，响应包含x个结果，如果有超过0的结果，我可以再次调用它以获得下一个x个结果。x可以任意选择直到API的最大页面大小。

现在，我想要在WebSocket上高效地流式传输全套结果，而不会使它不堪重负（施加背压）。最初，我构建了整个resultset，然后从中创建了一个源代码：

getEventsFuture().foreach { events =>
  sender ! Flow.fromSinkAndSource(Sink.ignore, Source(events))
}

这样可以工作，WebSocket客户机以其最大速度接收所有事件。这样做的最大缺点是，在开始向客户机返回数据之前，我必须获取所有页面。理想情况下，我会使用较小的页面大小，并在连接后立即将结果返回给客户机，获取该过程中的下一个页面。

因此，我需要一个流，我可以添加数据的源后，流已经物化。我尝试使用source.actorref:

val events = Source.actorRef[Event](1000, OverflowStrategy.fail).mapMaterializedValue { outActor =>
  sendEvents(outActor)
  NotUsed
}

sender ! Flow.fromSinkAndSource(Sink.ignore, events)

本质上，我使用物化的actorRef并将所有事件发送给它。每次提取页面时，我都会将结果转储给执行元。现在，我对源代码的初始化可能已经告诉您，这并不总是起作用。有时，当响应足够大，而客户端没有像其他时候那样迅速地消耗时，套接字连接就会关闭。我觉得overflowstrategy.fail是反对丢弃事件的正确策略，因为我不希望客户机认为他们得到了所有东西，如果不是这样的话。

我没有要预先为缓冲区设置的sane值，我不想设置int.max或其他东西，因为我认为Akka内部确实为缓冲区大小分配了全部内存。

我该怎么解决？我希望所有的事件尽可能快地和适当的背压到客户端，就像在第一个例子。

一旦第一个页面被取出，我就知道总共会有多少结果，所以我可以在前面取出一个小页面，并将缓冲区大小设置为完整的结果大小，但这似乎是一个变通方法。

Answer 1

我发现UnfoldAsync非常适合这个用例。

def unfoldAsync[S, E](s: S)(f: S ⇒ Future[Option[(S, E)]]): Source[E, NotUsed]

就像unfold，但是fold函数返回一个future，它将导致源代码完成或在源代码完成时发出。

当有需求时发出，且展开状态返回，未来完成时具有某些值

当展开函数返回的future以空值完成时完成