为 http 响应编写效果流

Question

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

TL;DR：我需要帮助发送效果 F 的结果和一个函数，return 应该每 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 是我的自定义案例 classes.The 问题是我在 timeOrErr <- times.get(city.toUpperCase) 行中只获得一次时间。因此，它每秒发送完全相同的值（例如 2:43:31、2:43:31 等。我希望它是 2:43:31、2:43:32 等）。而且我不知道如何让这个函数每秒都被调用。

此外，我尝试使用一种略有不同的技术（以及许多其他与此类似的技术）：

       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 等功能。如果我尝试 return Ok(stream)，那么 Circe 无法序列化 F 因为它是抽象的，所以它甚至没有被编译。

第三种方法是：

       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 几乎没有关于它的信息。我很高兴听到任何建议！

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

为 http 响应编写效果流

Compose stream of effects for http response

scala

scala-cats

http4s

fs2

cats-effect