如何用Circe从Http4s Response获取Stream[IO, Job]？

问题根源 #

你请求的是text/event-stream（SSE）类型响应，但默认的CirceEntityCodec未提供Stream[IO, Job]的EntityDecoder实例——SSE响应需要先解析为Server-Sent Event对象，再从中提取数据并解码为Job。

解决步骤 #

1. 添加http4s-sse依赖 #

在build.sbt中添加和你的http4s主版本一致的SSE支持模块：

"org.http4s" %% "http4s-sse" % "0.23.14"

2. 导入必要的编解码器 #

替换原有导入，补充SSE与circe结合的解码工具：

import io.circe.generic.auto._
import org.http4s.circe.CirceEntityCodec._
import org.http4s.sse._
import org.http4s.circe.decodeJson
// 其他必要导入

3. 修改响应解析逻辑 #

不要直接调用response.as[Stream[IO, Job]]，先将响应解析为SSE事件流，再解码每个事件的data字段为Job：

"should return the stream of all jobs" in {
  for {
    response <- jobsRoutes.orNotFound.run(
      Request(
        method = Method.GET,
        uri = uri"/jobs",
        headers = Headers(Accept(MediaType.`text/event-stream`))
      )
    )
    sseStream <- response.as[Stream[IO, ServerSentEvent]]
    jobStream = sseStream.flatMap { event =>
      event.data match {
        case Some(data) => Stream.eval(decodeJson[Job](data))
        case None => Stream.empty
      }
    }
    _ <- jobStream.compile.drain // 消耗流，验证解码无错误
  } yield {
    response.status shouldBe Status.Ok
  }
}

补充说明 #

• ServerSentEvent是http4s定义的SSE事件模型，包含data、event、id等字段，我们需要从data中提取JSON字符串并解码为Job。
• decodeJson[Job](data)会借助你已有的circe自动解码器，将字符串解析为Job实例，返回IO[Job]后通过Stream.eval转换为流元素。

内容的提问来源于stack exchange，提问作者riccardo.cardin