Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Delay.kt (Kotlin)

Code before:
 * produces the following emissions
 *
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout Timeout period
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration
): Flow<T> = timeoutInternal(timeout.toDelayMillis())

private fun <T> Flow<T>.timeoutInternal(
    timeoutMillis: Long
): Flow<T> = scopedFlow { downStream ->
    if (timeoutMillis <= 0L) throw FlowTimeoutException("Timed out immediately")

    // Produce the values using the default (rendezvous) channel
    // Similar to [debounceInternal]
    val values = produce<Any?> {
        var timeoutJob = launch { // Emits timeout unless cancelled
            delay(timeoutMillis)
            send(TIMEOUT)
        }
        try {
            collect {
                timeoutJob.cancel() // Upstream emitted, so cancel the job

                send(it ?: NULL)

                // We reset the job here. The reason being is that the `flow.emit()` suspends, which in turn suspends `send()`.
                // We only want to measure a timeout if the producer took longer than `timeoutMillis`, not producer + consumer
                timeoutJob = launch {
                    delay(timeoutMillis)
                    send(TIMEOUT)
                }
            }
        } finally {
            timeoutJob.cancel()
            send(DONE) // Special signal to let flow end
        }
    }

    // Await for values from our producer now
    whileSelect {
        values.onReceive { value ->
            if (value !== DONE) {

Review comment (performance):
Also, it seems like there exists both more straightforward, idiomatic and efficient solution (tho I haven't really tested it):

```
private fun <T> Flow<T>.timeoutInternal(
    timeoutMillis: Long
): Flow<T> = scopedFlow { downStream ->
    if (timeoutMillis <= 0L) throw FlowTimeoutException("Timed out immediately")
    val values = buffer(Channel.RENDEZVOUS).produceIn(this)
    whileSelect {
        values.onReceiveCatching { value ->
            value.onSuccess {
                downStream.emit(it)
            }.onClosed {
                return@onReceiveCatching false
            }
            return@onReceiveCatching true
        }
        onTimeout(timeoutMillis) {
            throw FlowTimeoutException(timeoutMillis)
        }
    }
}
```

Could you please take a look at it?

Code after:
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout period. If non-positive, the flow is timed out immediately
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration
): Flow<T> = timeoutInternal(timeout)

private fun <T> Flow<T>.timeoutInternal(
    timeout: Duration
): Flow<T> = scopedFlow { downStream ->
    if (timeout <= Duration.ZERO) throw TimeoutCancellationException("Timed out immediately")
    val values = buffer(Channel.RENDEZVOUS).produceIn(this)
    whileSelect {
        values.onReceiveCatching { value ->
            value.onSuccess {
                downStream.emit(it)
            }.onClosed {
                return@onReceiveCatching false
            }
            return@onReceiveCatching true
        }
        onTimeout(timeout) {
            throw TimeoutCancellationException("Timed out waiting for $timeout")
        }
    }
}