libraryDependencies ++= {
  val akkaV = "2.4.3"
  Seq(
    "com.typesafe.akka" %% "akka-stream" % akkaV,
    "com.typesafe.akka" %% "akka-http-experimental" % akkaV
  )
}

package mananan.eval.akka.stream

import akka.actor.ActorSystem
import akka.http.scaladsl._
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, StatusCode}
import akka.stream._
import akka.stream.scaladsl._

import scala.util.{Failure, Success, Try}

/**
  * リバースプロクシのサンプルプログラム
  * localhost:8080 で受けたリクエストをlocalhost:8081～8083に負荷分散する
  * 
  * Created by まななん on 2016/04/10.
  */
object ReverseProxySample extends App {
    implicit val system = ActorSystem("my-system")
    implicit val materializer = ActorMaterializer()

    // リバースプロキシからアクセスするエンドポイント一覧
    case class Endpoint(host: String, port: Int)
    val endpoints = for {
        host <- "localhost" :: Nil
        port <- 8081 to 8083
    } yield Endpoint(host, port)

    // サーバーリクエストをエンドポイントへのクライアントリクエストに変換するFlow(手抜き ※ゆえ、warnが出る)
    val convertReqFlow = Flow[HttpRequest].map { (_, None) }

    // Endpointの分だけrequestを渡すとresponseを戻すFlowをつくる
    val httpClientFlows = endpoints.map { case Endpoint(host, port) =>
        Http().cachedHostConnectionPool[None.type](host, port)
    }

    // エンドポイントからのクライアントレスポンスをサーバーレスポンスに変換するFlow(手抜き)
    val convertResFlow = Flow[(Try[HttpResponse], None.type)].map {
        case (Success(res), _) =>
            res
        case (Failure(e), _) =>
            e.printStackTrace()
            HttpResponse(StatusCode.int2StatusCode(500))
    }

    // リバースプロクシの動きをするFlow Graphをつくる
    val reverseProxyGraph = GraphDSL.create() { implicit builder =>
        import GraphDSL.Implicits._

        val input = builder.add(convertReqFlow)
        val balance = builder.add(Balance[(HttpRequest, None.type)](httpClientFlows.size))
        val merge = builder.add(Merge[(Try[HttpResponse], None.type)](httpClientFlows.size))
        val output = builder.add(convertResFlow)

        input ~> balance
        httpClientFlows.foreach { flow =>
            balance ~> flow ~> merge
        }
        merge ~> output

        FlowShape(input.in, output.out)
    }
    val reverseProxyFlow = Flow.fromGraph(reverseProxyGraph)

    // リバースプロクシを起てる   
    Http().bindAndHandle(reverseProxyFlow, "localhost", 8080)

    // 動作デモ用にエンドポイントをローカルホストに起てる(どのパスにきてもエンドポイントのポート番号を戻すだけ)
    endpoints.foreach { case Endpoint(host, port) =>
        Http().bindAndHandle(server.Directives.complete(s"${port}"), host, port)
    }
}

    val reverseProxyGraph = GraphDSL.create() { implicit builder =>
        import GraphDSL.Implicits._

        val input = builder.add(convertReqFlow)
        val partition = builder.add(Partition[(HttpRequest, None.type)](httpClientFlows.size, { case (req, _) =>
            req.uri.toRelative.toString match {
                case uri if uri.startsWith("/hoge") => 0
                case uri if uri.startsWith("/piyo") => 1
                case _ => 2
            }
        }))
        val merge = builder.add(Merge[(Try[HttpResponse], None.type)](httpClientFlows.size))
        val output = builder.add(convertResFlow)

        input ~> partition
        httpClientFlows.foreach { flow =>
            partition ~> flow ~> merge
        }
        merge ~> output

        FlowShape(input.in, output.out)
    }