def chainRequests(reqOption: Option[HttpRequest]): Future[Option[(Option[HttpRequest], HttpResponse)]] =
reqOption match {
case Some(req) => Http().singleRequest(req).flatMap { response =>
// handle the error case. Here we just return the errored response
// with no next item.
if (response.status.isFailure()) Future.successful(Some(None -> response))
// Otherwise, convert the response to a strict response by
// taking up the body and looking for a next request.
else convertToStrict(response).map { strictResponse =>
getNextRequest(strictResponse) match {
// If we have no next request, return Some containing an
// empty state, but the current value
case None => Some(None -> strictResponse)
// Otherwise, pass on the request...
case next => Some(next -> strictResponse)
}
}
}
// Finally, there's no next request, end the stream by
// returning none as the state.
case None => Future.successful(None)
}
2条答案
按热度按时间vktxenjb1#
如果你正在做一个网络爬虫,看看this post,这个答案解决了一个更简单的情况,比如下载分页的资源,其中到下一个页面的链接在当前页面响应的标题中。
您可以使用
Source.unfoldAsync
方法创建一个链式源-其中一个项目通向下一个项目。这需要一个函数,该函数接受元素S
并返回Future[Option[(S, E)]]
以确定流是否应该继续发出E
类型的元素,并将状态传递给下一次调用。在你的情况下,这有点像:
1.取初始值
HttpRequest
1.生成
Future[HttpResponse]
1.如果响应指向另一个URL,则返回
Some(request -> response)
,否则返回None
然而,有一个问题,如果流中不包含指向下一个请求的指针,它将不会从流中发出响应。
为了解决这个问题,你可以让传递给
unfoldAsync
的函数返回Future[Option[(Option[HttpRequest], HttpResponse)]]
。这允许你处理以下情况:下面是一些注解代码,概述了这种方法,但首先是初步的:
HttpEntity
从(潜在的)流转换为严格的实体:*字符串
接下来,几个函数从
HttpResponse
创建Option[HttpRequest]
。这个例子使用了类似Github的分页链接的方案,其中Links
头部包含,例如:<https://api.github.com/...> rel="next"
:型
接下来,我们将传递给
unfoldAsync
的真实的函数。它使用Akka HTTPHttp().singleRequest()
API获取HttpRequest
并生成Future[HttpResponse]
:型
请注意,如果我们得到一个错误响应,流将不会继续,因为我们在下一个状态中返回
None
。你可以调用它来获得一个
HttpResponse
对象流,如下所示:型
至于返回最后一个(或错误)响应的值,您只需使用
Sink.last
,因为流将在成功完成或第一个错误响应时结束。例如:型
n1bvdmb62#
可以使用
Source.unfoldAsync
。字符串
完整的源代码和可运行的项目可以在over on GitHub中找到