scala - 如何以编程方式关闭 fs2.StreamApp？

Question

扩展StreamApp要求您提供stream def。它有一个 requestShutdown 参数。

def Stream(args: List[String], requestShutdown: F[Unit]): Stream[F, ExitCode]


我提供了此方法的实现，并了解 args 是作为命令行参数传入的。不过，我不确定什么提供了 requestShutdown 参数以及我可以用它做什么。

具体来说，我想在正在启动 Http4s 服务器(永远阻塞)的 Stream[IO, ExitCode] 上调用正常关闭。

看起来需要一个Signal并且必须设置？我试图“了解”的底层流如下所示:

for {
   scheduler <- Scheduler[IO](corePoolSize = 1)
   exitCode  <- BlazeBuilder[IO]
                    .bindHttp(port, "0.0.0.0")
                    .mountService(services(scheduler), "/")
                    .serve
    } yield exitCode

我的stream def是here fs2 项目中的 StreamAppSpec 和 StreamAppSpec 中有一些内容。但我不知道如何适应它。

Answer 1

您可以将提供给流函数的 requestShutdown 参数视为执行时请求终止程序的操作。 p>

执行它会导致程序结束。

这是一个使用示例:

  override def stream(args: List[String], requestShutdown: IO[Unit]): Stream[IO, ExitCode] =
    for {
      scheduler <- Scheduler[IO](corePoolSize = 1)
      exitStream = scheduler.sleep[IO](10 seconds)
       .evalMap(_ => requestShutdown)
       .map(_ => ExitCode.Success)
      serverStream = BlazeBuilder[IO]
        .bindHttp(port, "0.0.0.0")
        .mountService(services(scheduler), "/")
        .serve
      result <- Stream.emits(List(exitStream, serverStream)).joinUnbounded
    } yield result

在这种情况下，我们创建两个流:

• 第一个会等待10秒才会触发效果
  终止应用程序。

• 第二个将运行 http4s 服务器。

然后我们加入这两个流，以便它们同时运行，这意味着 Web 服务器将运行 10 秒，然后另一个流发出信号表明程序应终止。