fs2 -在错误时恢复的循环流

Question

我试图创建一个fs2流，

stream

• Interrupts

• 每3秒计算一次函数，并在发生错误时将该数据保存到文件

• 中，必要时尝试基于单独的流验证和重新启动相同的

• 。

下面是我的方法，但我不确定这是否是最好的方法。

我使用大写的函数名来突出显示它被调用的位置。

private def CREATE_STREAM_RECURSIVE(
     session: AuthSession,
     myService: MyService[IO],
     sessionStore: SessionStore[IO],
  ): Stream[IO, Unit] = {
    Stream
      .repeatEval(service.fetchData(session))
      .flatMap { data =>
        val jsonStream = Stream.eval(IO.delay(data.asJson.noSpaces))
        persistToFile(jsonStream) <-- another fs2 stream that writes to a file
      }
      .metered(3.seconds)
      .handleErrorWith { _ =>
        Stream.force {
          myService.login()
            .flatMap(sessionStore.setSession)
            .map(CREATE_STREAM_RECURSIVE(_, myService, sessionStore)) //recursive
        }
      }
      .interruptWhen(isItOkToInterruptStream)
  }

问题：

• 如果继续发生错误，这种递归是否有可能导致堆栈崩溃?当中断发生时，该流是否有可能停止而不完成文件写入?如果是的话，是否可以确保该文件在退出发生之前被完全写入？

谢谢

Answer 1

在SystemFw的帮助下，我能够通过fs2不和谐渠道找到答案。

If an error keeps happening, is it possible for this recursion to blow the stack?

不，因为递归是一元的，它通过一个flatMap，在本例中是由force隐藏的，因此它是堆栈安全的

When the interruption happens, is it possible for this stream to stop without completing the file write?

是

If so, is there anyway to ensure that the file is written completely before the exit happens?

是的，通过将文件转换为IO并在其上调用.uncancelable，使其写入不可中断