https://kotlinlang.org logo
#coroutines
Title
# coroutines
d

df

11/26/2020, 5:01 PM
Hey there. Not really an issue but I'm trying to understand the behavior of the following flow.
Copy code
fun getContent(dropN: Long) = flow {
        val counter = AtomicLong(dropN)
        val resource = javaClass.getResource("/products.csv").toURI()

        File(resource).useLines { lines ->
            lines.filter { it.isNotEmpty() }
                .drop(dropN.toInt())
                .map { it.toLong() }
                .chunked(100)
                .onEach { counter.addAndGet(it.size.toLong()) }
                .map { getContentByProductIds(it) }
                .onEach { println("Processed ${counter.get()} lines from input file") }
                .forEach { emit(it) }
        }
    }
The flow is collected by multiple co-routines and I see exactly the same lines printed multiple times.
Copy code
Processed 100 lines from input file
Processed 100 lines from input file
Given the flow definition I would assume the second onEach should only be invoked when the first onEach was invoked before, which should always increment the counter. Why do I still see the same counter value multiple times?
t

tseisel

11/26/2020, 5:32 PM
Flow are cold: everytime a coroutine collects a flow, its block is executed again from the start. If you want multiple coroutines to share the same counter, consider using SharedFlow (via
shareIn
operator)
☝️ 1
d

Dominaezzz

11/26/2020, 6:47 PM
counter
is being reset for each flow#
a

Adam Powell

11/26/2020, 6:48 PM
A flow is like a function. If you have a function that prints something, it will print that thing once for each time you call it because it runs the whole function once for each time it is called. Collecting a flow runs the whole flow each time it is collected.
1
Each collect on a flow is isolated from every other collect, and the flow only executes when it is collected. This is what people mean when they say flows are, "cold" (by default, until you start adding other fancy things like SharedFlow or StateFlow)
d

df

11/26/2020, 6:51 PM
Okay now I guess it made click. What would be the correct approach if I want to implement single-producer-multiple-consumers (fan-out) with a producer that does not produce more than the consumers can handle?
thanks for the explanation already
the individual collectors should not process the same emitted value, but should distribute the emitted values.
d

Dominaezzz

11/26/2020, 6:57 PM
produceIn(...).receiveAsFlow()
. Recommend looking up docs for these functions.
d

df

11/26/2020, 6:59 PM
will do, thanks
2 Views