我正在尝试使用Akka Streams实现分页。
case class SomeObject(id:Long, next_page:Option[Map[String,String]])
def chainRequests(uri: Uri): Future[Option[(Uri, T)]] = {
if (uri.isEmpty) return Future.successful(None)
val response: Future[Response[T]] = sendWithRetry(prepareRequest(HttpMethods.GET, uri)).flatMap(unmarshal)
response.map { resp =>
resp.next_page match {
case Some(next_page) => Some(next_page("uri"), resp.data)
case _ => Some(Uri.Empty, resp.data)
}
}
}
Source.single(SomeObject).map(Uri(s"object/${_.id}")).map(uri => Source.unfoldAsync(url)(chainRequest)).map(...some processing goes here)
字符串
问题是,如果我执行source.take(1000),并且分页有很多元素(页面),那么下游在Source.unfoldAsync完成之前不会获得新元素。
我试着在流动中使用循环,
val in = builder.add(Flow[Uri])
val out = builder.add[Flow[T]]
val partition = b.add(Partition[Response[T]](2,r => r.next_page match {case Some(_)=>1; case None => 0}))
val merge = b.add(Merge[Response[T]],2)
in ~> mergeUri ~> sendRequest ~> partition
mergeUri.preferred <~ extractNextUri <~ partition.out(1)
partition.out(0) ~> Flow[Response[T]].map(_.data) ~> out
FlowShape(in.in, out.out)
型
但是上面的代码不起作用。
我一直在创建我自己的GraphStage。UnfoldAsync需要第一个元素,但在Flow解决方案中,我没有“第一个”元素。有什么建议吗?
谢谢
2条答案
按热度按时间qlfbtfca1#
通过编写自己的GraphStage找到了解决方案
字符串
evrscar22#
Source.unfoldAsync
就是你要找的。我准备了一个简单的项目,它遍历REST API的所有页面,并累积所有页面的结果,返回Future with Seq。
您可以在over on GitHub中找到完整的源代码和项目
字符串