ser*_*erg 2 scala akka akka-http
我不知道如何cachedHostConnectionPool使用 scala 在 akka-http 中创建发送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 = "/"))
Run Code Online (Sandbox Code Playgroud)
似乎scala版本只支持普通主机名,而在java中你也可以提供一个协议(来自他们的测试):
http.cachedHostConnectionPool("akka.io", materializer());
http.cachedHostConnectionPool("https://akka.io", materializer());
http.cachedHostConnectionPool("https://akka.io:8080", materializer());
Run Code Online (Sandbox Code Playgroud)
任何已知的解决方法?
您必须改用cachedHostConnectionPoolHttps:
val poolClientFlow = Http().cachedHostConnectionPoolHttps[Promise[HttpResponse]]("example.com")
Run Code Online (Sandbox Code Playgroud)
| 归档时间: |
|
| 查看次数: |
1090 次 |
| 最近记录: |