Route akka-http request through a proxy

Question

I am rewriting some application layer code in scala from using scalaj to akka-http in order to reduce the number of third party dependencies in the project (we already use akka for other things in the same project.) The code simply wraps common types of request to an underlying general request provided by the library

Mostly it has been fine, but I am stuck on the problem of optionally adding a proxy to a request.

Requests should either be direct to the destination or via a proxy, determined by a parameter at runtime.

In my scalaj implementation, I have the following helper class and methods

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)

Is there a way to build a similar construct with akka-http?

Answer 1

Akka HTTP does have proxy support that, as of version 10.0.9, is still unstable. Keeping in mind that the API could change, you could do something like the following to handle optional proxy settings:

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

In Akka Http 10.2.0, use bindflow for a Flow[HttpRequest, HttpResponse, NotUsed] defined by a RunnableGraph with Flowshape. Insided the RunnableGraph, an Http() outgoingConnection is used to connect to the remote proxy. Some example code:

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