produce and Pipelines

produce

produce { } starts a coroutine that sends into a channel and returns it as a ReceiveChannel. The channel is closed automatically when the block ends. Chaining producers gives a pipeline.

numbers() ─► onlyEven() ─► square() ─► consumer
    1,2,3...     2,4,6...      4,16,36...

Pipeline Example

Example

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.numbers(upTo: Int): ReceiveChannel<Int> = produce {
    for (x in 1..upTo) send(x)
}

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.onlyEven(input: ReceiveChannel<Int>) = produce {
    for (x in input) if (x % 2 == 0) send(x)
}

@OptIn(ExperimentalCoroutinesApi::class)
fun CoroutineScope.square(input: ReceiveChannel<Int>) = produce {
    for (x in input) send(x * x)
}

fun main() = runBlocking {
    val result = square(onlyEven(numbers(10)))
    result.consumeEach { println(it) }
    println("Pipeline done")
}

Output

4
16
36
64
100
Pipeline done

Note: For most data-processing pipelines a Flow is simpler and safer (cold, no manual closing). Channels shine when coroutines really need to hand work to each other.