通过代理路由 akka-http 请求

Question

我正在 scala 中重写一些应用程序层代码，从使用 scalaj 到 akka-http 为了减少项目中第三方依赖的数量（我们已经在同一个项目中将akka用于其他事情。）代码只是将常见类型的请求包装到库提供的底层通用请求中

大部分情况下都很好，但我被困在可选地向请求中添加代理的问题上。

请求应该直接到达目的地或通过代理，由运行时的参数确定。

在我的 scalaj 实现中，我有以下帮助程序类和方法

object HttpUtils {
  private def request(
               host: Host,
               method: HttpMethod,
               params: Map[String, String],
               postData: Option[String],
               timeout: Duration,
               headers: Seq[(String, String)],
               proxy: Option[ProxyConfig]
             ): HttpResponse[String] = {
    // most general request builder. Other methods in the object fill in parameters and wrap this in a Future
    val baseRequest = Http(host.url)
    val proxiedRequest = addProxy(proxy, baseRequest)
    val fullRequest = addPostData(postData)(proxiedRequest)
      .method(method.toString)
      .params(params)
      .headers(headers)
      .option(HttpOptions.connTimeout(timeout.toMillis.toInt))
      .option(HttpOptions.readTimeout(timeout.toMillis.toInt))
    fullRequest.asString  // scalaj for send off request and block until response
  }

      // Other methods ...

   private def addProxy(proxy: Option[ProxyConfig], request: HttpRequest): HttpRequest =
     proxy.fold(request)((p: ProxyConfig) => request.proxy(p.host, p.port))
}

case class ProxyConfig(host: String, port: Int)

有没有办法用 akka-http 构建类似的结构？

Answer 1

Akka HTTP 确实有 proxy support，从 10.0.9 版开始，它仍然不稳定。请记住，API 可能会更改，您可以执行以下操作来处理可选代理设置：

import java.net.InetSocketAddress

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.http.scaladsl.{ClientTransport, Http}

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()

case class ProxyConfig(host: String, port: Int)

val proxyConfig = Option(ProxyConfig("localhost", 8888))
val clientTransport =
  proxyConfig.map(p => ClientTransport.httpsProxy(InetSocketAddress.createUnresolved(p.host, p.port)))
             .getOrElse(ClientTransport.TCP)

val settings = ConnectionPoolSettings(system).withTransport(clientTransport)
Http().singleRequest(HttpRequest(uri = "https://google.com"), settings = settings)

Answer 2

在 Akka Http 10.2.0 中，对带有 Flowshape 的 RunnableGraph 定义的 Flow[HttpRequest, HttpResponse, NotUsed] 使用 bindflow。在 RunnableGraph 内部，一个 Http() 传出连接用于连接到远程代理。一些示例代码：

import akka.actor.typed.ActorSystem
import akka.actor.typed.scaladsl.Behaviors
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream._
import akka.stream.scaladsl.{Broadcast, Flow, GraphDSL, Merge}

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration.DurationInt
import scala.io.StdIn
import scala.util.{Failure, Success}

object Main {

  def main(args: Array[String]) {

    implicit val system: ActorSystem[Nothing] = ActorSystem(Behaviors.empty, "testproxy")
    implicit val executionContext: ExecutionContextExecutor = system.executionContext
    system.log.info("TestAkkaHttpProxy Main started...")
    val remoteHost = "xxx.xxx.xxx.x"
    val remotePort = 8000
    val proxyHost = "0.0.0.0"
    val proxyPort = 8080

    val gateway = Flow.fromGraph(GraphDSL.create() { implicit b =>
      import GraphDSL.Implicits._

      // Broadcast for flow input
      val broadcaster = b.add(Broadcast[HttpRequest](1))
      // Merge for flow output
      val responseMerge = b.add(Merge[HttpResponse](1))
      // outgoing client for remote proxy
      val remote = Http().outgoingConnection(remoteHost, remotePort)
      // filter out header that creates Akka Http warning
      val requestConvert = Flow[HttpRequest]
        .map(req => { req.mapHeaders(headers => headers.filter(h => h.isNot("timeout-access")))
        })
      // connect graph
      broadcaster.out(0) ~> requestConvert ~> remote ~> responseMerge
      // expose ports
      FlowShape(broadcaster.in, responseMerge.out)
    })

    // Akka Http server that binds to Flow (for remote proxy)
    Http().newServerAt(proxyHost, proxyPort).bindFlow(gateway)
      .onComplete({
        case Success(binding) ⇒
          println(s"Server is listening on 0.0.0.0:8080")
          binding.addToCoordinatedShutdown(hardTerminationDeadline = 10.seconds)
        case Failure(e) ⇒
          println(s"Binding failed with ${e.getMessage}")
          system.terminate()
      })

    system.log.info("Press RETURN to stop...")
    StdIn.readLine()
    system.terminate()
  }
}