Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Kotlin flow buffer capacity

Tags:

kotlin

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?

like image 723
Pedro Alipio Avatar asked Jul 26 '26 08:07

Pedro Alipio


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

like image 86
marstran Avatar answered Jul 29 '26 04:07

marstran


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
  • ...
like image 44
Mafor Avatar answered Jul 29 '26 03:07

Mafor



Donate For Us

If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!