For a given flow, is there a way to emit an initia...

# coroutines

[Mark]

For a given flow, is there a way to emit an initial value should the first value not be emitted before some timeout?

[Mark]

My use case for this is when using combine where you don’t want an input flow to potentially forever hold up the combine flow

[Mark]

fun <T> Flow<T>.emitInitialValueOnTimeout(
    timeoutMillis: Long,
    initialValue: T
): Flow<T> = flow {
    try {
        withTimeout(timeoutMillis) {
            emit([e-mail]())
        }
        emitAll([e-mail](1))
    } catch (e: TimeoutCancellationException) {
        emit(initialValue)
        emitAll(this@emitInitialValueOnTimeout)
    }
}

https://pl.kotl.in/IpOwN6OK3
But the first(), drop(1) doesn’t look safe to me

[Zach Klippenstein (he/him) [MOD]]

You want to apply this downstream of the combine?

[Zach Klippenstein (he/him) [MOD]]

Your impl assumes the upstream is a hot flow, which might not be true generally

[Mark]

I would apply it to each input of combine.

[Mark]

maybe the receiver flow should be converted to a SharedFlow first?

[Zach Klippenstein (he/him) [MOD]]

This should work: https://pl.kotl.in/K7EUWBvDI

fun <T> Flow<T>.emitInitialAfterTimeout(
    timeoutMillis: Long,
    initialValue: T
): Flow<T> = channelFlow {
    var fallbackJob: Job? = launch {
        delay(timeoutMillis)
        send(initialValue)
    }

    collect {
        fallbackJob?.let {
            it.cancelAndJoin()
	        fallbackJob = null
        }
        send(it)
    }
}

[Mark]

Yep, yours looks much nicer. Thanks

[Mark]

Suppose I want to reuse the code for something like:

fun <T> Flow<T>.emitInitialNullOnTimeout(
    timeoutMillis: Long,
): Flow<T?>

but is there any way to specify a type R as T or T? so that can return Flow<R>?

fun <T, R: T or T?> Flow<T>.emitInitialValueOnTimeout(
    timeoutMillis: Long,
    initialValue: R,
): Flow<R>

[Mark]

Here’s a nice variation, useful for emitting a cached value if that can be obtained before the first value:

fun <T> Flow<T>.emitInitialValueIfBeforeFirst(
    initialValue: suspend () -> T,
): Flow<T> = channelFlow {
    var initialValueJob: Job? = launch {
        val value = initialValue()
        logi("emitting initialValue: $value because first value not yet emitted")
        send(value)
    }
    collect { item ->
        initialValueJob?.let { job ->
            job.cancelAndJoin()
            initialValueJob = null
        }
        send(item)
    }
}

which means can also use it for the earlier extension fun:

fun <T> Flow<T>.emitInitialValueOnTimeout(
    timeoutMillis: Long,
    initialValue: T,
): Flow<T> = emitInitialValueIfBeforeFirst {
    delay(timeoutMillis)
    initialValue
}

[reline]

for the nullable case it should infer the nullability already but you could make another extension function for convenience

Untitled.kt

[Mark]

Oh yes, of course! Also means probably better to just use emitInitialAfterTimeout() directly.