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

Code before:
 * if it gets cancelled while waiting for the next element,
 * or if [MutableCollection.add] fails with an exception.
 * In both cases, the exception will be used for cancelling the channel and then rethrown.
 *
 * The intended use case for this function is collecting the remaining elements of a closed channel
 * and processing them in a single batch.
 *
 * Example:
 * ```
 * val doContinue = AtomicBoolean(true)
 *
 * // Start the sender
 * val channel = produce {
 *     var i = 0
 *     while (doContinue.load()) {
 *         send(i++)
 *         delay(10.milliseconds)
 *         if (i == 42) break
 *     }
 * }
 *
 * // Start the consumer
 * launch {
 *     // Read elements until we suddenly decide to stop
 *     // or until the channel is closed.
 *     while (Random.nextInt() % 100 != 0) {
 *         val nextElement = channel.receiveCatching()
 *         if (nextElement.isClosed) return@launch
 *         println("Received ${nextElement.getOrNull()}")
 *     }
 *     delay(100.milliseconds)
 *     doContinue.store(false)
 *     delay(100.milliseconds)
 *     val remainingElements = mutableListOf<Int>()
 *     try {
 *         channel.consumeTo(remainingElements)
 *     } finally {
 *         println("Remaining elements: $remainingElements")
 *     }
 * }
 * ```
 */
public suspend fun <E, C: MutableCollection<E>> ReceiveChannel<E>.consumeTo(collection: C): C =
    consumeEach(collection::add).let { collection }


@PublishedApi
internal fun ReceiveChannel<*>.cancelConsumed(cause: Throwable?) {
    cancel(cause?.let {
        it as? CancellationException ?: CancellationException("Channel was consumed, consumer had failed", it)
    })

Review comment (suggestion):
It's better to teach users to use even adjacent things properly. `Random.nextInt(100)` requires less random bits per invokation and is also genuinely uniformly distributed.

Code after:
 * Example:
 * ```
 * val doContinue = AtomicBoolean(true)
 *
 * // Start the sender
 * val channel = produce {
 *     var i = 0
 *     while (doContinue.load()) {
 *         send(i++)
 *         delay(10.milliseconds)
 *         if (i == 42) break
 *     }
 * }
 *
 * // Start the consumer
 * launch {
 *     // Read elements until we suddenly decide to stop
 *     // or until the channel is closed.
 *     while (Random.nextInt(100) != 0) {
 *         val nextElement = channel.receiveCatching()
 *         if (nextElement.isClosed) return@launch
 *         println("Received ${nextElement.getOrNull()}")
 *     }
 *     delay(100.milliseconds)
 *     doContinue.store(false)
 *     delay(100.milliseconds)
 *     val remainingElements = mutableListOf<Int>()
 *     try {
 *         channel.consumeTo(remainingElements)
 *     } finally {
 *         println("Remaining elements: $remainingElements")
 *     }
 * }
 * ```
 */
public suspend fun <E, C: MutableCollection<E>> ReceiveChannel<E>.consumeTo(collection: C): C =
    consumeEach(collection::add).let { collection }


@PublishedApi
internal fun ReceiveChannel<*>.cancelConsumed(cause: Throwable?) {
    cancel(cause?.let {
        it as? CancellationException ?: CancellationException("Channel was consumed, consumer had failed", it)
    })
}