Hands-On Design Patterns with Kotlin

Establishing a pipeline

Now, to establish our pipeline:

val pagesProducer = producePages()val domProducer = produceDom(pagesProducer)val titleProducer = produceTitles(domProducer)runBlocking {    titleProducer.consumeEach {        println(it)    }}

We have the following:

pagesProducer |> domProducer |> titleProducer |> output

A pipeline is a great way to break a long process into smaller steps. Note that each producing coroutine is a pure function, so it's also easy to test and reason about.

The entire pipeline could be stopped by calling cancel() on the first coroutine in line.

We can achieve an even nicer API by using the extension functions:

private fun ReceiveChannel<Document>.titles(): ReceiveChannel<String> {    val channel = this    fun getTitles(dom: ...