Repository: ktorio/ktor
File: ktor-io/common/src/io/ktor/utils/io/ByteReadChannelOperations.kt (Kotlin)

Code before:
@OptIn(InternalAPI::class, InternalIoApi::class)
public fun ByteReadChannel.readAvailable(min: Int, block: (Buffer) -> Int): Int {
    require(min > 0) { "min should be positive" }
    require(min <= CHANNEL_MAX_SIZE) { "Min($min) shouldn't be greater than $CHANNEL_MAX_SIZE" }

    if (availableForRead < min) return -1
    return block(readBuffer.buffer)
}

public class ReaderScope(
    public val channel: ByteReadChannel,
    override val coroutineContext: CoroutineContext
) : CoroutineScope

public class ReaderJob internal constructor(
    public val channel: ByteWriteChannel,
    public override val job: Job
) : ChannelJob {
    /**
     * To avoid the risk of concurrent write operations, we cancel the nested job before
     * performing `flushAndClose` on the [ByteWriteChannel].
     */
    public suspend fun flushAndClose() {
        job.cancelChildren()
        job.children.toList().joinAll()
        channel.flushAndClose()
    }
}

@Suppress("UNUSED_PARAMETER")
public fun CoroutineScope.reader(
    coroutineContext: CoroutineContext = EmptyCoroutineContext,
    autoFlush: Boolean = false,
    block: suspend ReaderScope.() -> Unit
): ReaderJob = reader(coroutineContext, ByteChannel(), block)

@OptIn(InternalCoroutinesApi::class)
public fun CoroutineScope.reader(
    coroutineContext: CoroutineContext,
    channel: ByteChannel,
    block: suspend ReaderScope.() -> Unit
): ReaderJob {
    val job = launch(coroutineContext) {
        val nested = Job(this.coroutineContext.job)
        try {
            block(ReaderScope(channel, this.coroutineContext + nested))
            nested.complete()

            if (this.coroutineContext.job.isCancelled) {
                channel.cancel(this.coroutineContext.job.getCancellationException())
            }

Review comment (suggestion):
Minor. We could avoid an extra list creation.
```suggestion
        job.children.forEach { it.join() }
```

Code after:
@OptIn(InternalAPI::class, InternalIoApi::class)
public fun ByteReadChannel.readAvailable(min: Int, block: (Buffer) -> Int): Int {
    require(min > 0) { "min should be positive" }
    require(min <= CHANNEL_MAX_SIZE) { "Min($min) shouldn't be greater than $CHANNEL_MAX_SIZE" }

    if (availableForRead < min) return -1
    return block(readBuffer.buffer)
}

public class ReaderScope(
    public val channel: ByteReadChannel,
    override val coroutineContext: CoroutineContext
) : CoroutineScope

public class ReaderJob internal constructor(
    public val channel: ByteWriteChannel,
    public override val job: Job
) : ChannelJob {
    /**
     * To avoid the risk of concurrent write operations, we cancel the nested job before
     * performing `flushAndClose` on the [ByteWriteChannel].
     */
    @InternalAPI
    public suspend fun flushAndClose() {
        job.cancelChildren()
        job.children.forEach { it.join() }
        channel.flushAndClose()
    }
}

@Suppress("UNUSED_PARAMETER")
public fun CoroutineScope.reader(
    coroutineContext: CoroutineContext = EmptyCoroutineContext,
    autoFlush: Boolean = false,
    block: suspend ReaderScope.() -> Unit
): ReaderJob = reader(coroutineContext, ByteChannel(), block)

@OptIn(InternalCoroutinesApi::class)
public fun CoroutineScope.reader(
    coroutineContext: CoroutineContext,
    channel: ByteChannel,
    block: suspend ReaderScope.() -> Unit
): ReaderJob {
    val job = launch(coroutineContext) {
        val nested = Job(this.coroutineContext.job)
        try {
            block(ReaderScope(channel, this.coroutineContext + nested))
            nested.complete()

            if (this.coroutineContext.job.isCancelled) {
                channel.cancel(this.coroutineContext.job.getCancellationException())