Add withTimeout combinator

[Z1kkurat]

On a day-to-day job basis we often use such a combinator for Stream, allowing us to time out and interrupt streams that do not complete within given FiniteDuration. Such a combinator may be useful as a part of the Stream itself

[Z1kkurat]

It's strange that 2.12 build has failed, b/c the failure is in the suite that was not touched

[kubukoz]

Isn't this just interruptAfter(n.seconds)?

[Z1kkurat]

@kubukoz seems like interruptAfter just interrupts (stops) stream, not resulting in any error if Stream wasn't completed yet, whereas this combinator will result in TimeoutException on evaluation if stream was not completed within given timeout

[kubukoz]

It won't result in any error indeed, but you can write a timeout operator on top of interruptAfter by checking the exit case, and reusing most of the logic of that (you'll probably need a Deferred with an optional TimeoutException that you'll only complete with the exception in onFinalizeCase if the case is Canceled. I can give you a snippet of what I meant later

[kubukoz]

I meant something like this:

def timeout[F[_]: Concurrent: Timer, A](duration: FiniteDuration): Pipe[F, A, A] = in =>
  Stream.eval(Deferred[F, Either[TimeoutException, Unit]])
    .flatMap { deff =>
      in.onFinalizeCase {
        case ExitCase.Canceled => deff.complete(Left(new TimeoutException(s"Timed out after $duration")))
        case _                 => deff.complete(Right(()))
      }.interruptAfter(duration) ++ Stream.eval_(deff.get.rethrow)
    }

[Z1kkurat]

@kubukoz Thanks! Yes, a solution like that came to my mind, but I thought that we wouldn't be able to distinguish between user-requested cancellation and one caused by the timeout. Am I wrong?

[kubukoz]

Oh, I suppose that might be true, really good point.

How about this...

def timeout[F[_]: Concurrent: Timer, A](duration: FiniteDuration): Pipe[F, A, A] = 
  _.interruptWhen(Stream.sleep[F](duration).as(true))

I don't know why I didn't come up with that sooner.

Edit: no, wait, that's too simple. here we go again...

[Z1kkurat]

This will not result in an error, I guess? That will just interrupt the stream. Maybe something like that will do?

def timeout[F[_]: Concurrent: Timer, A](duration: FiniteDuration): Pipe[F, A, A] = 
  _.interruptWhen(Timer[F].sleep(duration).as(Left(new TimeoutException(s"Timed out after $duration"))))

[kubukoz]

Yeah, that should do it 😅

[Z1kkurat]

What's your opinion on what this operator should be - a function on a Stream or a Pipe? I guess that's just a matter of usage at call-site - Stream.emit(...).timeout(...) vs Stream.emit(...).through(timeout)

[kubukoz]

I think it should be a function on Stream. Such is the convention for these, and we don't really have a good place to put pipes that are that generic in.

[Z1kkurat]

Thank you very much for advice! The implementation became much shorter and clearer 😅

[kubukoz]

I think we can use context bounds here

[kubukoz]

Would you mind also testing composability of timeouts? i.e. if you do (s1.timeout(d1) ++ s2).timeout(d2), what happens depending on the times of s1, s2, d1 > d2 (if d2 > d1, s1 might make it in time but s2 might not, so we still want to timeout, etc.)

[kubukoz]

Just a couple minor things, and I think we're good to go. I'd also like to see what @SystemFw thinks.

[Z1kkurat]

I added context bounds and a couple of tests as you recommended. It seems like other test cases (with different s1 and s2) are specific sub-cases of these ones and are derived from the composability of Stream. If you have any more test cases in mind - please let me know.

[kubukoz]

Sure, it makes sense. Thanks a lot!