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

Question

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

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

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

具体来说，我想在Stream[IO, ExitCode]启动Http4s服务器（永久阻止）的上调用正常关机。

看起来Signal是必需的，必须设置吗？我试图“了解”的基础流如下所示：

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

我的streamdef在这里，并且StreamAppSpec来自fs2项目中有一些内容，StreamAppSpec但是我无法弄清楚如何适应它。

Answer 1

您可以认为requestShutdown提供给流函数的参数是指一个动作，该动作在执行时将要求终止程序。

因此，执行它会导致它结束程序。

这是一个示例用法：

  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秒钟。