Akka Http Proxy Transport (no CONNECT).

AkkaHttpProxyTransport.scala

package com.chatmeter.common.util.http

import java.net.InetSocketAddress

import akka.actor.ActorSystem
import akka.http.scaladsl.model.headers.{HttpCredentials, `Proxy-Authorization`}
import akka.http.scaladsl.settings.ClientConnectionSettings
import akka.http.scaladsl.{ClientTransport, Http}
import akka.stream.scaladsl.{BidiFlow, Flow, Keep}
import akka.util.ByteString

import scala.concurrent.Future

case class AkkaHttpProxyTransport(
  proxyAddress: InetSocketAddress,
  proxyCredentials: Option[HttpCredentials],
  underlyingTransport: ClientTransport = ClientTransport.TCP,
  debug: Boolean = true
) extends ClientTransport {

  override def connectTo(
    host: String, port: Int,
    settings: ClientConnectionSettings
  )(implicit system: ActorSystem): Flow[ByteString, ByteString, Future[Http.OutgoingConnection]] = {


    BidiFlow.fromFlows(
      rewriteOutgoingForProxy(host, port).map(printDebug("> ")),
      Flow.apply[ByteString].map(printDebug("< "))
    )
      .joinMat(
        underlyingTransport.connectTo(proxyAddress.getHostString, proxyAddress.getPort, settings)
      )(Keep.right)

      // on the HTTP level we want to see the final remote address in the `OutgoingConnection`
      .mapMaterializedValue(_.map(_.copy(remoteAddress = InetSocketAddress.createUnresolved(host, port)))(system.dispatcher))
  }


  /**
    * Rewrites the http request meta to conform to the proxy format
    *
    * before:
    *
    *   GET / HTTP/1.1
    *   ...
    *
    * after:
    *
    * GET http://icanhazip.com:80/ HTTP/1.1
    * Proxy-Connection: Keep-Alive
    * Proxy-Authorization: Basic cGF1bGdob3N0OmNtODM1Ng==
    * ...
    *
    */
  private def rewriteOutgoingForProxy(host: String, port: Int): Flow[ByteString, ByteString, _] = Flow.fromFunction(byteString => {

    val (lead, trail) = byteString.span(_ != '\n'.toByte)
    val withFullPath = lead.utf8String.replace(" /", " " + "http" + "://" + host + ":" + port + "/")
    val withProxyHeaders = ByteString(
      withFullPath +
      "\nProxy-Connection: Keep-Alive" +
      proxyCredentials.map(creds => s"\n${`Proxy-Authorization`(creds)}").getOrElse("")
    )
    withProxyHeaders.concat(trail)
  })

  private def printDebug(prefix: String)(byteString: ByteString): ByteString = if (debug) {
    println(prefix + byteString.utf8String)
    byteString
  } else byteString

}

Use it:

import akka.actor.ActorSystem
import akka.http.scaladsl.model.HttpRequest
import akka.http.scaladsl.model.headers.{BasicHttpCredentials, HttpCredentials}
import akka.http.scaladsl.settings.ConnectionPoolSettings
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import com.chatmeter.common.util.http.AkkaHttpProxyTransport
import scala.concurrent.Await
import scala.concurrent.duration.Duration

val proxyHost: String = ???
val proxyPort: Int = ???
val auth: Option[HttpCredentials] = Some(
   BasicHttpCredentials("username", "password")
)
implicit val sys = ActorSystem()
implicit val ec = sys.dispatcher

// only http supported (does not form CONNECT tunnel, you can use a ClientTransport.httpsProxy for that
val req = HttpRequest(uri = "http://icanhazip.com") 
val transport = AkkaHttpProxyTransport(
  InetSocketAddress.createUnresolved(proxyHost, proxyPort),
  proxyCredentials = auth
)
val settings = ConnectionPoolSettings(actorSystem).withTransport(transport)
Await.result(for {
  resp <- akka.http.scaladsl.Http().singleRequest(req, settings = settings)
  entity <- Unmarshal(resp.entity).to[String]
} yield {
  println("ip is " + entity)
}, Duration.Inf)

Thanks for the solution. I've run into problems when using transfer-encoding: chunked. I think the flow was executed multiple times (per chunk), therefore leading to crazy results :)

The solution to this was to add some state to rewriteOutgoingForProxy, which rewrites only the first time.