scala - 超时后停止 fs2-stream

Question

我想使用类似于 take(n: Int) 的函数但在时间维度上:consume(period: Duration .因此，如果发生超时，我希望流终止。我知道我可以将流编译为类似 IO[List[T]] 的内容。并取消它，但我会失去结果。实际上，我想将无尽的流转换为有限的流并保留结果。

更多关于更广泛的问题。我有来自消息代理的源源不断的事件流，但我也有轮换凭据来连接到代理。所以我想要的是消耗事件流一段时间，然后停止，获取新凭据，再次连接到创建新流的代理并将两个流连接成一个。

Answer 1

有一种方法可以做到这一点:

/**
    * Interrupts this stream after the specified duration has passed.
    */
  def interruptAfter[F2[x] >: F[x]: Concurrent: Timer](duration: FiniteDuration): Stream[F2, O]