将Akka-http-client请求链接到流中

Question

我想使用akka-http-client作为流来链式http请求。链中的每个http请求都依赖于前一个请求的成功/响应，并使用它来构造一个新的请求。如果一个请求不成功，流应该返回不成功请求的响应。

如何在AKKA-HTTP中构造这样的流？我应该使用哪一个akka-http客户端级别的API？

Answer 1

如果你正在做一个网络爬虫，看看这个帖子。这个答案处理一个更简单的情况，例如下载分页资源，其中到下一页的链接在当前页响应的一个标题中。

您可以使用source.UnfoldAsync方法创建一个链接源-其中一个项指向下一个项。这将接受一个函数，该函数接受元素S并返回future[option[(S,E)]]以确定流是否应继续发射E类型的元素，并将状态传递给下一次调用。

在你的例子中，这有点像：

1. 获取初始httpRequest
2. 生成将来[HttpResponse]
3. 如果响应指向另一个URL，则返回some（请求->响应），否则返回none

但是，有一个问题，即如果流不包含指向下一个请求的指针，则不会从流中发出响应。

null

import scala.concurrent.duration._

def convertToStrict(r: HttpResponse): Future[HttpResponse] =
  r.entity.toStrict(10.minutes).map(e => r.withEntity(e))

接下来，使用几个函数从httpResponse创建选项[HttpRequest]。此示例使用类似GitHub的分页链接的方案，其中links头包含，例如: rel=“next” :

def nextUri(r: HttpResponse): Seq[Uri] = for {
  linkHeader <- r.header[Link].toSeq
  value <- linkHeader.values
  params <- value.params if params.key == "rel" && params.value() == "next"
} yield value.uri

def getNextRequest(r: HttpResponse): Option[HttpRequest] =
  nextUri(r).headOption.map(next => HttpRequest(HttpMethods.GET, next))

接下来，我们将传递给UnfoldAsync的真实函数。它使用Akka HTTPHTTP().SingleRequest()API获取HttpRequest并生成Future[HttpResponse]:

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)
  }

注意，如果我们得到错误的响应，流将不会继续，因为我们在下一个状态中返回none。

您可以调用此命令来获取httpresponse对象流，如下所示：

val initialRequest = HttpRequest(HttpMethods.GET, "http://www.my-url.com")
Source.unfoldAsync[Option[HttpRequest], HttpResponse](
    Some(initialRequest)(chainRequests)

至于返回最后一个（或错误的）响应的值，您只需使用sink.last，因为流将在成功完成或在第一个错误的响应时结束。例如：

def getStatus: Future[StatusCode] = Source.unfoldAsync[Option[HttpRequest], HttpResponse](
      Some(initialRequest))(chainRequests)
    .map(_.status)
    .runWith(Sink.last)