Kotlin flow buffer capacity

user_7709729 0
Viewed 2394

I have a question about Kotlin flow buffer capacity. The following code:

import kotlinx.coroutines.flow.*

suspend fun main() = coroutineScope {
    flow { 
        for (i in 1..3) { 
            println("Emiting $i")
            emit(i) 
        }
    }.buffer(0)
    .collect { 
        value -> 
            delay(100)
            println("Consuming $value")
    }
}

generates the following output:

Emiting 1
Emiting 2
Consuming 1
Emiting 3
Consuming 2
Consuming 3

If I remove the buffer, the result is:

Emiting 1
Consuming 1
Emiting 2
Consuming 2
Emiting 3
Consuming 3

Shall I assume that when the capacity is 0 means that is actually 1?

kotlin
2 Answers

No, the capacity is actually 0. It just looks like you are buffering an element because the collect is first consuming an element, and then delays for 100ms. That allows the flow to emit another element in the meanwhile.

The buffer function actually creates a second coroutine that allows the flow and collect functions to execute concurrently. Without the call to buffer, each element must be done with the collect before flow can continue with the next element, because both functions are executed by the same coroutine.

Let's step through your code to figure out how it happens:

  1. The flow delays for 1ms, prints Emitting 1, emits 1, and suspends.
  2. Collect immediately consumes 1, then starts the delay for 100ms.
  3. The flow continues because 1 was consumed, delays for another 1ms, prints Emitting 2, emits 2, then suspends.
  4. Collect finishes the delay of 100ms, prints Consuming 1, then consumes 2 and delays for another 100ms.
  5. The flow continues because 2 was consumed, delays for another 1ms, prints Emitting 3, emits 3, then suspends.
  6. Collect finishes the delay of 100ms, prints Consuming 2, then consumes 3 and delays for another 100ms.
  7. Collect finishes the delay of 100ms, prints Consuming 3, then finishes collecting.
  8. The flow continues, and finishes because there's no more elements to emit.

You can read more about buffer here: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/buffer.html

user_41374890

Not really, it's 0. The buffer(0) method uses a channel under the hood, and size 0 effectively makes the channel an unbuffered channel. The flow and consumer both need to be ready to emit/consume:

Unbuffered channels transfer elements when sender and receiver meet each other (aka rendezvous). buffered-channels

  • Flow prints "Emiting 1"
  • Flow tries to emit 1 and suspends (no buffer, nor reader awaiting)
  • Collector is ready for reading. Flow emits 1, collector reads 1.
  • Collector suspends on delay 100
  • Flow prints "Emiting 2"
  • Flow tries to emit 2 and suspends (no buffer, nor reader awaiting)
  • Collector prints "Consuming 1"
  • Collector is ready for reading. Flow emits 2, collector reads 2
  • Collector suspends on delay 100
  • Flow prints "Emiting 3"
  • Flow tries to emit 3 and suspends (no buffer, nor reader awaiting)
  • Collector prints "Consuming 2"
  • Collector is ready for reading. Flow emits 3, collector reads 3
  • Collector suspends on delay 100
  • Flow exits
  • Collector prints "Consuming 3" and exits

When you remove the buffer(0), the situation is a bit different. there is no channel involved. The flow is a "cold-stream", i.e. it kind of waits for the consumer to pull next items:

  • Collector "subscribes" to the Flow
  • Flow prints "Emiting 1", emits "1" and suspends
  • Collector reads "1", suspends for 100ms and prints "Consuming 1"
  • Collector asks for the next item
  • Flow prints "Emiting 2", emits "2" and suspends
  • ...
user_90907510
Related