为 http 响应编写效果流

问题描述

我想通过 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:31 等。我希望它是 2:43:31、2:43:32 等）。而且我不知道如何让这个函数每秒被调用一次。

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)
         }
       }

这里的问题是嵌套单子的地狱。 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)。

不幸的是，官方 docs 几乎没有关于它的信息。我很高兴听到任何建议！

解决方法

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

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)))
  }
}