Hello! :hand: I need a timeout mechanism for the f...

# flow

[galex]

Hello! ✋
I need a timeout mechanism for the first item emitted only but .timeout() is called after each emit, so I made the following function:

/**
 * If the first element takes too long to emit, the [onTimeout] callback will be called without disturbing the current flow
 */
fun <T> Flow<T>.timeoutFirst(duration: Long, scope: CoroutineScope, onTimeout: () -> Unit): Flow<T> = flow {
    
    var emitted = false

    scope.launch {
        delay(duration)
        if (!emitted) {
            onTimeout()
        }
    }

    collect {
        emitted = true
        emit(it)
    }
}

First question:

Is there a way to get the coroutineScope where the flow will be collected so that I don't need to pass a scope?
I was using coroutineScope {} before instead of scope.launch but then I couldn't get it to work in unit tests under runTest

Second question:

Instead of a callback, could I use the throwing mechanism as timeout does?

Thanks in advance! 😊

[franztesca]

fun <T> Flow<T>.timeoutFirst(duration: Duration) = flow {
    coroutineScope {
        val values = buffer(Channel.RENDEZVOUS).produceIn(this)
        withTimeout(duration) {
            emit(values.receiveCatching().onClosed { return@withTimeout }.getOrThrow())
        }
        emitAll(values.consumeAsFlow())
    }
}

[aarkling]

OMG I literally came here for this exact same problem 😀

[aarkling]

What is the buffer for?

[aarkling]

I think this works (modification of alex’s example):

class TimeoutFirstCancellationException : CancellationException()

fun <T> Flow<T>.timeoutFirst(duration: Duration): Flow<T> = flow {
  coroutineScope {
    var emitted = false

    launch {
      delay(duration)
      if (!emitted) {
        throw TimeoutFirstCancellationException()
      }
    }

    collect {
      emitted = true
      emit(it)
    }
  }
}

[aarkling]

Nvm. Above doesn’t work.

[aarkling]

This does:

private fun <T> Flow<T>.timeoutFirst(duration: Duration): Flow<T> = channelFlow {
    val emitted = Job()

    launch {
      collect {
        emitted.cancelAndJoin()
        send(it)
      }
    }

    try {
      withTimeout(duration) { emitted.join() }
    } catch (e: TimeoutCancellationException) {
      if(emitted.isActive) throw e
    }
  }

[galex]

@aarkling I ended up with this:

fun <T : Any?> Flow<T>.timeoutFirstNotNull(duration: Long, onTimeout: () -> Unit): Flow<T> = flow {
    coroutineScope {

        var emitted = false

        launch {
            delay(duration)
            if (!emitted) {
                onTimeout()
            }
        }

        collect {
            if (it != null && !emitted) {
                emitted = true
            }
            emit(it)
        }
    }
}

[galex]

That's without throw mechanism, yours looks actually better I think