Hi there, how to add a timeout to

awaitClose

on

CallbackFlow

? (details in thread)

to transform a callback into a flow

I’d ask if transforming the callback into regular coroutine wouldn’t be enough in your case? In most cases cancellation would be handled on collector side

withTimeout { foo().collect { } }

, but I assume you precisely want to expose self-timeouting Flow, right? If so, I’d suggest to add custom

timeout()

extension function. If I understood correctly what you want to achieve is simply:

private fun <T> Flow<T>.timeout() = channelFlow {
    withTimeout(3000) {
        collect { send(it) }
    }
}

which would complete successfully if the inner flow completes within 3 seconds, fail with TimeoutException otherwise Unless I misunderstood your intentions 🤔

Thanks for the feedback @mateusz.kwiecinski!

but I assume you precisely want to expose self-timeouting Flow, right?

Yes, that’s correct 😉.

If so, I’d* *suggest to add custom 

timeout()

 extension function.

This approach works, however has a little downside that I don’t have access to the original scope. The following example is a bit closer to the actual code. Notice the

cancelable

part.

fun foo() {
        return callbackFlow {
            val cancelable=abc.method1(object : Callback {
                ... // non-terminal methods
                ... // terminal methods
            })
            try {
                withTimeout(3000) {
                    awaitClose()
                }
            } catch (e: CancellationException) {
                cancelable.cancel()
            }
        }
    }

----------------- I tried to create a custom “awaitCloseWithTimeout” and it seems to work as expected. Having said that it’s not exactly a beatiful code, so any suggestions are more than appreciated 😉

fun foo() {
        return callbackFlow {
            val cancelable=abc.method1(object : Callback {
                ... // non-terminal methods
                ... // terminal methods
            })
            
            awaitCloseWithTimeout(TIMEOUT) { success ->
                if (!success) cancelable.cancel()
            }
        }
    }


private suspend fun ProducerScope<*>.awaitCloseWithTimeout(timeout: Long, block: suspend (Boolean) -> Unit) {
        check(kotlin.coroutines.coroutineContext[Job] === this) { "awaitClose() can only be invoked from the producer context" }
        var success = false
        try {
            withTimeoutOrNull(timeout) {
                suspendCancellableCoroutine<Unit> { cont ->
                    invokeOnClose {
                        success = true
                        cont.resume(Unit)
                    }
                }
            }
        } finally {
            block(success)
        }
    }

I'm working on adding something like this to the coroutines library. If you want to try implementing something custom though, this is what I used externally: https://gist.github.com/pablobaxter/50cb9873f38bf6a0e2d72c769ac112f3