为http响应编写效果流

Question

我想通过HTTP进行流式JSON响应。其目的是每秒发送给定城市的当前时间。

TL；DR：我需要帮助发送效果为F的结果，并且应该每隔1秒调用一个返回此效果的函数。你知道有什么简单的方法吗

我已经尝试了几种方法，但都不起作用。

1. 这是第一个：

def timeStreamingRoutes[F[_]: Sync : Timer](times: Times[F]): HttpRoutes[F] = {
val dsl = new Http4sDsl[F]{}
import dsl._

HttpRoutes.of[F] {
case GET -> Root / "streaming" / city =>

val throttling = Stream.awakeEvery[F](1.second)

for {
timeOrErr <- times.get(city.toUpperCase)
resp <- timeOrErr match {
case Right(time) => Ok(throttling.map(_ => time.asJson))
case Left(error) => BadRequest(throttling.map(_ => error.asJson))
}
} yield resp
}
}

在这里，我的times.get(city.toUpperCase)函数具有以下签名：

def get(city: String): F[Either[Times.CurrentTimeError, Times.CurrentTime]]

，其中CurrentTimeError和CurrentTime是我的自定义事例类。问题是我在timeOrErr <- times.get(city.toUpperCase)行只得到一次时间。所以，它每秒钟都会发送完全相同的值(比如，2:43:31，2:43:30，等等。我希望它是2:43:3，2:43:12，等等(。我不知道如何让这个函数每秒钟都被调用。

2. 此外，我尝试使用一种稍微不同的技术(以及许多其他类似的技术(：

def timeStreamingRoutes[F[_]: Sync : Timer](times: Times[F]): HttpRoutes[F] = {
val dsl = new Http4sDsl[F]{}
import dsl._

HttpRoutes.of[F] {
case GET -> Root / "streaming" / city =>

val throttling = Stream.awakeEvery[F](1.second)
val payload    = Stream(
for {
timeOrErr <- times.get(city.toUpperCase)
resp      <- timeOrErr match {
case Right(time) => Ok(time.asJson)
case Left(error) => BadRequest(error.asJson)
}
} yield resp
)

val stream = throttling.zipRight(payload)

Ok(stream)
}
}

这里的问题是嵌套monad的地狱。stream具有Stream[F, F[Response[F]]]类型。我不能使它成为一个合适的F[Response[F]]，因为fs2-Stream不提供像sequence或traverse这样的函数。如果我试图返回Ok(stream)，那么Circe无法序列化F，因为它是抽象的，所以它甚至没有被编译。

3. 第三种方法是：

HttpRoutes.of[F] {
case GET -> Root / "streaming" / city =>

val throttling = Stream.awakeEvery[F](1.second)
val payload    = Stream(
for {
timeOrErr <- times.get(city.toUpperCase)
resp      <- timeOrErr match {
case Right(time) => time.asJson
case Left(error) => error.asJson
}
} yield resp
)

val stream = throttling.map(_ => payload)

Ok(stream)
}

好吧，数字3也没有被编译。主要是因为我不能在payload中编写monad。也就是说，case Right(time) => time.asJson和case Left(error) => error.asJson必须类似于case Right(time) => SomethingThatcanBeUsedAsALastWrapperInThisForComprehension(time.asJson)。

不幸的是，官方文档中几乎没有关于它的信息。我很乐意听到任何建议！

Answer 1

据我所知，你想要这样的东西：

import cats.effect.{ContextShift, IO, Timer}
import fs2.Stream
import io.circe.Encoder
import io.circe.generic.semiauto.deriveEncoder
import org.http4s.{EntityEncoder, HttpRoutes}
import org.http4s.circe.streamJsonArrayEncoderOf
import org.http4s.dsl.Http4sDsl
import scala.concurrent.duration._ // Provides the second extension method.
type Error = String
final case class Time(data: Long)
trait Times {
def get(cityName: String): IO[Either[Error, Time]]
}
object MyService extends Http4sDsl[IO] {
// JSON Encoders.
private implicit final val CirceEncoder: Encoder[Time] =
deriveEncoder
private implicit final val timeOrErrorCirceEncoder: Encoder[Either[Error, Time]] =
Encoder.either[Error, Time](leftKey = "error", rightKey = "time")
private implicit final val timeOrErrorEntityEncoder: EntityEncoder[IO, Stream[IO, Either[Error, Time]] =
streamJsonArrayEncoderOf

/** Returns the service itself. */
def apply(times: Times)
(implicit ev1: ContextShift[IO], ev2: Timer[IO]):HttpRoutes[IO] = HttpRoutes.of[IO] {
Root / "streaming" / city =>
Ok(Stream.awakeEvery[IO](1.second).evalMap(_ => times.get(city.toUpperCase)))
}
}