scala - 如何在 scala 中为 https 请求创建 akka-http cachedHostConnectionPool？

Question

我不知道如何使用 scala 在 akka-http 中创建 cachedHostConnectionPool 来发送 https 请求。 queueRequest(HttpRequest(uri = "https://example.com") 向http发送请求，cachedHostConnectionPool[Promise[HttpResponse]]("https://example.com") 抛出 : 不是预期字符的错误。

import scala.util.{ Failure, Success }
import scala.concurrent.{ Future, Promise }

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._

import akka.stream.{ OverflowStrategy, QueueOfferResult }

implicit val system = ActorSystem()
import system.dispatcher // to get an implicit ExecutionContext into scope
implicit val materializer = ActorMaterializer()

val QueueSize = 10

// This idea came initially from this blog post:
// http://kazuhiro.github.io/scala/akka/akka-http/akka-streams/2016/01/31/connection-pooling-with-akka-http-and-source-queue.html
val poolClientFlow = Http().cachedHostConnectionPool[Promise[HttpResponse]]("example.com")
val queue =
  Source.queue[(HttpRequest, Promise[HttpResponse])](QueueSize, OverflowStrategy.dropNew)
    .via(poolClientFlow)
    .toMat(Sink.foreach({
      case ((Success(resp), p)) => p.success(resp)
      case ((Failure(e), p))    => p.failure(e)
    }))(Keep.left)
    .run()

def queueRequest(request: HttpRequest): Future[HttpResponse] = {
  val responsePromise = Promise[HttpResponse]()
  queue.offer(request -> responsePromise).flatMap {
    case QueueOfferResult.Enqueued    => responsePromise.future
    case QueueOfferResult.Dropped     => Future.failed(new RuntimeException("Queue overflowed. Try again later."))
    case QueueOfferResult.Failure(ex) => Future.failed(ex)
    case QueueOfferResult.QueueClosed => Future.failed(new RuntimeException("Queue was closed (pool shut down) while running the request. Try again later."))
  }
}

val responseFuture: Future[HttpResponse] = queueRequest(HttpRequest(uri = "/"))

似乎 scala 版本只支持纯主机名，而在 java 中你也可以提供协议(protocol)(来自他们的测试):

http.cachedHostConnectionPool("akka.io", materializer());
http.cachedHostConnectionPool("https://akka.io", materializer());
http.cachedHostConnectionPool("https://akka.io:8080", materializer());

有任何已知的解决方法吗？

Answer 1

你必须使用 cachedHostConnectionPoolHttps相反:

val poolClientFlow = Http().cachedHostConnectionPoolHttps[Promise[HttpResponse]]("example.com")