Hello, got a question on coroutines Flow. Is there a way of replicating RxJava `timeout` operator on...

# coroutines

[Doru N.]

Hello, got a question on coroutines Flow.
Is there a way of replicating RxJava timeout operator on Kotlin Flow? Namely, for my use case, I want to emit values from a Flow downstream, until a timeout happened (timeout duration should reset after every new emission), then stop emitting (no error) ?
The closest impl I see to what I need is the debounce operator, with the diff that I need to close / complete the flow when timeout happens.

[ephemient]

something like this? https://pl.kotl.in/n9-b2I_7X

[Doru N.]

thank you, this seems like it does the trick. one question: why did you had scope be as a parameter? (would creating a coroutineScope {} inside this function be more appropriate?)

[ephemient]

can't call coroutineScope {} inside the function, it's not in the right context

[ephemient]

kotlinx.coroutines internally has scopedFlow {} but it's not public

[baxter]

I saw a similar question on reddit some months back, and came up with this solution: https://gist.github.com/pablobaxter/50cb9873f38bf6a0e2d72c769ac112f3

[baxter]

This will timeout if the upstream takes too long to produce an item, but doesn't account for the time it takes to emit the same item downstream.

[baxter]

Basically will look like:

flow {
    emit("bar")
    delay(100)
    emit("baz")
    delay(100)
    emit("domo")
    delay(1000)
    emit("foo")
}.withTimeout(300).collect {
    delay(500)
    println(it)
}

The output would be:

> "bar"
> "baz"
> "domo"
> TimeoutException