Fan-in with Kotlin Coroutines and Flow

Sometimes you need to start several pieces of work at the same time and handle each result as soon as it is available. The fan-in pattern is a good fit for this: many concurrent producers send their results into one stream, and one consumer collects that stream.

Consider comparison shopping for a pair of headphones. If you want prices from four stores, you would not ask the first store, wait for its response, and only then ask the second store. You would send all four requests and process the quotes in the order they arrive.

In Kotlin, channelFlow gives us a convenient way to express that pattern while keeping the result as an ordinary Flow.

The example uses the kotlinx-coroutines-core dependency:

implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")

Here is the complete example:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.Locale
import kotlin.random.Random

data class PriceQuote(val store: String, val price: Double)

// Simulates a store API with its own latency
fun fetchPrice(store: String, product: String): Flow<PriceQuote> = flow {
    println("Requesting $product price from $store")
    delay(Random.nextLong(300, 2_000))
    emit(PriceQuote(store, Random.nextDouble(80.0, 120.0)))
}

fun comparePrices(product: String): Flow<PriceQuote> = channelFlow {
    val stores = listOf("Amazon", "Takealot", "eBay", "AliExpress")

    for (store in stores) {
        launch {
            fetchPrice(store, product).collect { quote ->
                send(quote)
            }
        }
    }
}

fun main() = runBlocking {
    comparePrices("headphones").collect { quote ->
        println("${quote.store}: \$${"%.2f".format(Locale.US, quote.price)}")
    }

    println("All stores answered.")
}

One run may produce the following output:

Requesting headphones price from Amazon
Requesting headphones price from Takealot
Requesting headphones price from eBay
Requesting headphones price from AliExpress
eBay: $94.13
Amazon: $101.87
AliExpress: $83.42
Takealot: $109.55
All stores answered.

The request is logged before each coroutine reaches delay, so those messages appear before their corresponding quotes. Neither the request messages nor the quotes have a guaranteed order, although the requests will normally start in store-list order in this small example.

Why use channelFlow instead of flow?

A normal flow { } is designed for sequential emission from its own coroutine. It preserves the coroutine context in which it is collected, so starting child coroutines inside it and calling emit from those children violates the flow invariant.

channelFlow is designed for the case where several coroutines need to produce values for the same flow. Its block provides a ProducerScope, and each child coroutine can safely call send on that scope. The channel merges those values into a single flow for the collector.

The resulting flow is still cold. Nothing happens when comparePrices("headphones") is called; the store requests start when the flow is collected. Collecting it a second time starts four new requests and produces a new set of quotes.

Why launch inside the loop?

Without launch, the collections would run one after another:

for (store in stores) {
    fetchPrice(store, product).collect { quote ->
        send(quote)
    }
}

collect is a suspending call and does not return until that store’s flow completes. Amazon would therefore finish before Takealot starts, followed by eBay and AliExpress. The total waiting time would be close to the sum of all four delays.

With launch, each store flow is collected by a child coroutine. All four delays can overlap, so the total waiting time is close to the slowest request rather than the sum of every request. This is concurrency; it does not require four dedicated threads. While one coroutine is suspended in delay or waiting for non-blocking I/O, another coroutine can make progress.

Why does the flow wait for every store?

The child coroutines launched inside channelFlow belong to its scope. Because this follows structured concurrency, the flow does not complete until the builder block and all of its children have completed.

The loop itself finishes quickly, but its children are still fetching and sending quotes. Only after the last child finishes does collection complete and All stores answered. print. There is no need to maintain a counter, close a channel manually, or call joinAll.

Structured concurrency also defines the failure behaviour. If one store flow throws an exception, its failure cancels the channelFlow and the other child coroutines, and the exception reaches the collector. If a failed store should not stop the comparison, handle that failure inside the individual store flow and decide what result to send instead.

Why use send instead of emit?

send is the channel operation provided by the ProducerScope inside channelFlow. Each store coroutine sends its quote into the same channel, and the downstream collector receives quotes as they become available. That interleaving of several producers into one consumer is the fan-in.

From main’s point of view, comparePrices is simply a Flow<PriceQuote>. The concurrency is kept inside the producer, so callers can use normal flow operators and terminal operations without knowing how the values were produced.

Cancellation and backpressure

Cancellation follows the same parent-child relationship. If the collector is cancelled or stops early, the work inside channelFlow and its child coroutines is cancelled as well. For example, comparePrices("headphones").first() returns the first quote to arrive and cancels the remaining requests. It finds the fastest response, not necessarily the cheapest price.

channelFlow uses a buffered channel by default. Producers can get ahead of a slow collector until that buffer is full; after that, send suspends until capacity becomes available. That gives the flow backpressure without allowing producers to grow an unbounded queue. If every send must wait for the collector, apply buffer(0) to the returned flow to use rendezvous behaviour.

The shorter version with merge

For this example, the merge flow operator is the idiomatic shortcut:

import kotlinx.coroutines.flow.merge

fun comparePrices(product: String): Flow<PriceQuote> {
    val stores = listOf("Amazon", "Takealot", "eBay", "AliExpress")

    return stores
        .map { store -> fetchPrice(store, product) }
        .merge()
}

merge collects all of the flows concurrently and emits their values without preserving the order of the source flows. For a straightforward fan-in, I would normally use merge. When the producer needs more control over how child coroutines send values, handle callbacks, or combine different kinds of concurrent work, channelFlow makes the underlying pattern explicit.

The useful part of fan-in is that the consumer does not have to coordinate the producers.

Each producer works at its own pace, while the collector sees one flow of results and retains the normal completion, failure, cancellation, and backpressure behaviour of Kotlin coroutines.