Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/channels/Channels.common.kt (Kotlin)

Code before:
 * Example:
 *
 * ```
 * suspend fun <E> ReceiveChannel<E>.consumeFirst(): E =
 *     consume { return receive() }
 * fun Int.cleanup() { println("cleaning up $this") }
 * val channel = Channel<Int>(10, onUndeliveredElement = Int::cleanup)
 * // Launch a procedure that creates values
 * launch(Dispatchers.Default) {
 *     repeat(10) {
 *         val sendResult = channel.trySend(it)
 *         if (sendResult.isFailure) {
 *             print("in the producer: ")
 *             it.cleanup()
 *         }
 *         yield()
 *     }
 * }
 * // Grab the first value and discard everything else
 * launch(Dispatchers.Default) {
 *     val firstElement = channel.consumeFirst()
 *     println("received $firstElement")
 * }
 * ```
 *
 * In this example, all ten values created by the producer coroutine will be processed: one by `consumeFirst`,
 * and the other ones by `Int.cleanup`, invoked either by [ReceiveChannel.cancel] inside [consume] or by the
 * producer itself when it observes failure.
 * In any case, exactly nine elements will go through a cleanup in this example.
 * If `consumeFirst` is implemented as `for (e in this) { return e }` instead, the cleanup does not happen.
 */
public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    var cause: Throwable? = null
    try {
        return block()
    } catch (e: Throwable) {
        cause = e
        throw e
    } finally {
        cancelConsumed(cause)
    }
}

/**
 * Performs the given [action] for each received element and [cancels][ReceiveChannel.cancel] the channel afterward.
 *
 * This function stops processing elements when the channel is [closed][SendChannel.close],
 * the coroutine in which the collection is performed gets cancelled and there are no readily available elements in the

Review comment (refactor):
Reading this, it seems like the example is mixing two concepts: one is about consuming, and the other about undelivered clean up (which we don't consider as "processed", it has a different meaning).

Maybe something more straightforward?
Like the following (outline, haven't tried to compile it):
```
val producer = produce {
    var currentElement = 0
    try {
        while(true) {
            send(currentElement++)
        }
    } finally {
        println("Cancelled after: $currentElement")
    }
}

consumeFirst { ... }
```

It's deterministic, more trivial and straightforward. WDYT?

Code after:
 *
 * Example:
 * ```
 * suspend fun <E> ReceiveChannel<E>.consumeFirst(): E =
 *    consume { return receive() }
 * // Launch a coroutine that constantly sends new values
 * val channel = produce(Dispatchers.Default) {
 *     var i = 0
 *     while (true) {
 *         // Will fail with a `CancellationException`
 *         // after `consumeFirst` finishes.
 *         send(i++)
 *     }
 * }
 * // Grab the first value and discard everything else
 * val firstElement = channel.consumeFirst()
 * check(firstElement == 0)
 * // *Note*: some elements could be lost in the channel!
 * ```
 *
 * In this example, the channel will get closed, and the producer coroutine will finish its work after the first
 * element is obtained.
 * If `consumeFirst` was implemented as `for (e in this) { return e }` instead, the producer coroutine would be active
 * until it was cancelled some other way.
 *
 * [consume] does not guarantee that new elements will not enter the channel after [block] finishes executing, so
 * some channel elements may be lost.
 * Use the `onUndeliveredElement` parameter of a manually created [Channel] to define what should happen with these
 * elements during [ReceiveChannel.cancel].
 */
public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    var cause: Throwable? = null
    try {
        return block()
    } catch (e: Throwable) {
        cause = e
        throw e
    } finally {
        cancelConsumed(cause)
    }
}

/**
 * Performs the given [action] for each received element and [cancels][ReceiveChannel.cancel] the channel afterward.
 *
 * This function stops processing elements when either the channel is [closed][SendChannel.close],
 * the coroutine in which the collection is performed gets cancelled and there are no readily available elements in the
 * channel's buffer,