如何使用 http4s 服务器和客户端库作为代理？

Question

我想使用 http4s 作为代理（如 nginx），如何将所有数据从我的 http4s 服务器转发到另一个 http 服务器？

我真正想做的是在执行转发功能之前在每个请求上附加一个验证功能。希望是这样的：

HttpService[IO] {
  case request =>
    val httpClient: Client[IO] = Http1Client[IO]().unsafeRunSync
    if(verifySuccess(request)) { // forward all http data to host2 and
                                 // get a http response.
      val result = httpClient.forward(request, "http://host2")
      result
    } else {
      Forbidden //403
    }
}

如何使用 http4s 和它的客户端来做到这一点？
谢谢

更新

在@TheInnerLight 的帮助下，我用sn-p 代码试了一下：

  val httpClient = Http1Client[IO]()

  val service: HttpService[IO] = HttpService[IO] {
    case req =>
      if(true) {
        for {
          client <- httpClient
          newAuthority = req.uri.authority.map(_.copy(host = RegName("scala-lang.org"), port = Some(80)))
          proxiedReq = req.withUri(req.uri.copy(authority = newAuthority))
          response <- client.fetch(proxiedReq)(IO.pure(_))
        } yield response
      } else {
        Forbidden("Some forbidden message...")
      }

  }

有一个请求：http://localhost:28080(http4s server listen at 28080):
但发生错误：

[ERROR] org.http4s.client.PoolManager:102 - Error establishing client connection for key RequestKey(Scheme(http),localhost) 
java.net.ConnectException: Connection refused
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.finishConnect(UnixAsynchronousSocketChannelImpl.java:252)
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.finish(UnixAsynchronousSocketChannelImpl.java:198)
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.onEvent(UnixAsynchronousSocketChannelImpl.java:213)
    at sun.nio.ch.KQueuePort$EventHandlerTask.run(KQueuePort.java:301)
    at java.lang.Thread.run(Thread.java:748)
[ERROR] org.http4s.server.service-errors:88 - Error servicing request: GET / from 0:0:0:0:0:0:0:1 
java.net.ConnectException: Connection refused
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.finishConnect(UnixAsynchronousSocketChannelImpl.java:252)
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.finish(UnixAsynchronousSocketChannelImpl.java:198)
    at sun.nio.ch.UnixAsynchronousSocketChannelImpl.onEvent(UnixAsynchronousSocketChannelImpl.java:213)
    at sun.nio.ch.KQueuePort$EventHandlerTask.run(KQueuePort.java:301)
    at java.lang.Thread.run(Thread.java:748)

最新版本

 val httpClient: IO[Client[IO]] = Http1Client[IO]()

  override val service: HttpService[IO] = HttpService[IO] {
    case req =>
      val hostName = "scala-lang.org"
      val myPort = 80
      if(true) {
        val newHeaders = {
          val filterHeader = req.headers.filterNot{h =>
            h.name == CaseInsensitiveString("Connection") ||
            h.name == CaseInsensitiveString("Keep-Alive") ||
            h.name == CaseInsensitiveString("Proxy-Authenticate") ||
            h.name == CaseInsensitiveString("Proxy-Authorization") ||
            h.name == CaseInsensitiveString("TE") ||
            h.name == CaseInsensitiveString("Trailer") ||
            h.name == CaseInsensitiveString("Transfer-Encoding") ||
            h.name == CaseInsensitiveString("Upgrade")
          }
          filterHeader.put(Header("host", hostName))
        }

        for {
          client <- httpClient
          newAuthority = req.uri.authority
            .map(_.copy(host = RegName(hostName), port = Some(myPort)))
            .getOrElse( Authority(host = RegName(hostName), port = Some(myPort)))
          proxiedReq = req.withUri(req.uri.copy(authority = Some(newAuthority)))
            .withHeaders(newHeaders)
          response <- client.fetch(proxiedReq)(x => IO.pure(x))
        } yield {
          val rst = response
          rst
        }

      } else {
        Forbidden("Some forbidden message...")
      }

  }

它对我的 REST API 网络服务器来说已经足够好了。
代理scala-lang.org进行测试时出现错误：

[ERROR] org.http4s.blaze.pipeline.Stage:226 - Error writing body 
org.http4s.InvalidBodyException: Received premature EOF.

Answer 1

这样的事情怎么样：

HttpService[IO] {
  case req =>
    if(verifyRequest(req)) {
      for {
        client <- Http1Client[IO]()
        newHost = "host2"
        newAuthority = Authority(host = RegName("host2"), port = Some(80))
        proxiedReq =
          req.withUri(req.uri.copy(authority = Some(newAuthority)))
           .withHeaders(req.headers.put(Header("host", newHost)))
        response <- client.fetch(proxiedReq)(IO.pure(_))
      } yield response
    } else {
      Forbidden("Some forbidden message...")
    }
}

请注意，您绝对应该避免在调用unsafeRunSync 时乱扔代码。您通常应该在您的程序中最多使用一次它（在Main 中）。在其他情况下，您应该专注于将效果提升到您正在工作的 monad 中。