从回调函数发送 zio http 响应

Question

我正在尝试使用他们的 simples hello world 示例来玩弄 ZIO http。我有一个 Java 编写的服务，它执行一些逻辑，它需要一个处理函数，所以它可以在结果准备好时调用它。如何将它与 ZIO http 一起使用？ 我想要这样的东西：

object HelloWorld extends App {

  def app(service: JavaService) = Http.collect[Request] {
    case Method.GET -> Root / "text" => {
      service.doSomeStuffWIthCallback((s:String) => Response.text(s))
    }
  }
  override def run(args: List[String]): URIO[zio.ZEnv, ExitCode] =
    Server.start(8090, app(new JavaService)).exitCode
}

基本上我想从回调函数发送 ZIO HTTP 响应，我只是不知道该怎么做。谢谢。

编辑：

我无法从您的代码中获得正确的类型，所以我决定简化整个事情，并得出以下结论：

val content: HttpData[Blocking, Throwable] = HttpData.fromStream {
    ZStream.fromEffect(doSomeStuffWrapped)
  }

  def doSomeStuffWrapped = {
    UIO.effectAsync[String] { cb =>
        cb(
          IO.succeed("TEST STRING")
        )
    }
  }

但是，这里的问题是类型不匹配，HttpData.fromStream需要字节的ZStream

这是我的要点的链接： https://gist.github.com/pmkyl/a37ff8b49e013c4e2e6f8ab5ad83e258

Answer 1

您应该使用effectAsync 将带有回调的Java 服务包装起来：

def doSomeStuffWrapped(service: JavaService): Task[String] = {
  IO.effectAsync[Throwable, String] { cb =>
    service.doSomeStuffWithCallback((s: String) => {
      // Success case
      cb(IO.succeed(s))
      // Optional error case?
      // cb(IO.fail(someException))
    })
  }
}

def app(service: JavaService) = Http.collectM[Request] {
  case Method.GET -> Root / "text" => {
    doSomeStuffWrapped(service)
      .fold(err => {
        // Handle errors in some way
        Response.text("An error occured")
      }, successStr => {
        Response.text(successStr)
      })
  }
}

您可能希望看到这篇介绍在 ZIO 中包装不纯代码的不同选项的文章：https://medium.com/@ghostdogpr/wrapping-impure-code-with-zio-9265c219e2e

Answer 2

这也是实现相同结果的另一种方法：

  case class MyService(name: String) {
    def imDone[R, E](s: String => Unit): Unit = s(name)
  }
  val s: MyService = MyService("test")
  val app: Http[Any, Nothing, Request, UResponse] = Http.collectM[Request] { case Method.GET -> Root / "text" =>
    ZIO.effectAsync[Any, Nothing, UResponse] { cb =>
      s.imDone { b =>
        cb(IO.succeed(Response.text(b)))
      }
    }
  }

Answer 3

在 ZIO-http v1.0.0.0-RC18 HttpData.fromStream 也可以使用 ZStream[R, E, String] 作为输入，Http 字符集默认为 CharsetUtil.UTF_8 但是您可以将任何字符集作为第二个参数传递给 HttpData.fromStream .您可以在下面找到解决方案

val stream: ZStream[Any, Nothing, String] = ZStream.fromEffect(doSomeStuffWrapped)
  val content: HttpData[Any, Nothing]       = HttpData.fromStream(stream)

  def doSomeStuffWrapped = {
    UIO.effectAsync[String] { cb =>
      cb(
        IO.succeed("TEST STRING"),
      )
    }
  }
  // Create HTTP route
  val app                = Http.collect[Request] {
    case Method.GET -> !! / "health" => Response.ok
    case Method.GET -> !! / "file"   => Response(data = content)
  }

  // Run it like any simple app
  override def run(args: List[String]): URIO[zio.ZEnv, ExitCode] =
    Server.start(8090, app.silent).exitCode
  

但是在以前的版本中，您可以执行如下所示的操作来使其正常工作

val stream: ZStream[Any, Nothing, Byte] =
    ZStream.fromEffect(doSomeStuffWrapped).mapChunks(_.map(x => Chunk.fromArray(x.getBytes(HTTP_CHARSET))).flatten)
  val content: HttpData[Any, Nothing]     = HttpData.fromStream(stream)

  def doSomeStuffWrapped = {
    UIO.effectAsync[String] { cb =>
      cb(
        IO.succeed("TEST STRING"),
      )
    }
  }
  // Create HTTP route
  val app                = Http.collect[Request] {
    case Method.GET -> !! / "health" => Response.ok
    case Method.GET -> !! / "file"   => Response(data = content)
  }

  // Run it like any simple app
  override def run(args: List[String]): URIO[zio.ZEnv, ExitCode] =
    Server.start(8090, app.silent).exitCode