// f1 is a flow of somewhat random numbers, no deeper meaning
    val f1 : Flow<Int> = flowOf(1, 1, 2, 3, 5, 8, 3, 1, 4, 5, 9, 4, 3, 7, 0, 7, 7, 4, 1, 5)
    /*
    Let's imagine that we are told f1 consists of "packets", which start with a length field followed by as many numbers
    as indicated in the length field. We want to transform f1 into a flow where each element is the sum of the numbers
    in the "payload" of one packet.
    */
    runBlocking {
        println("package sums:")
        f1.transformImperatively<Int, Int> {
            while (true) {
                if (hasNext()) {
                    emit(sum(next()))
                } else {
                    break
                }
            }
        }.collect(::println)
    }

suspend fun (FlowIterator<Int>).sum( count : Int ) : Int {
    var s : Int = 0
    for (i in 1..count) {
        if( hasNext() ) {
            s += next()
        } else {
            break
        }
    }
    return s
}

val flow: Flow<T> = TODO()
val channel = flow.produceIn(this@coroutineScope)
val iterator = channel.iterator()
iterator.hasNext()
iterator.next()

flow {
    val channel = inputFlow.produceIn(this@coroutineScope)
    val iterator = channel.iterator()
    while (iterator.hasNext()) {
        emit(transform(iterator.next()))
    }
}

.buffer(0)

produce { theFlow.collect { send(it) }  }

channelBasedTransformImperatively

correctTransformImperatively

theFlow.channelBasedTransformImperatively { ... } == theFlow.buffer(0).correctTransformImperatively { ... }

channelBasedTransformImperatively

.buffer(N)

theFlow.buffer(0) != theFlow

transformImperatively

inputFlow.buffer(0).map(transform)

suspend fun <T,R> Flow<T>.channelBasedTransformImperatively( transform : FlowContextAndIterator<T,R> ) {
  val inputFlow = this
  flow {
    val flowContext = this
    val channel = inputFlow.produceIn(this@coroutineScope)
    val iterator = channel.iterator()
    FlowContextAndIterator(flowContext, iterator).transform()
  }
}

FlowContextAndIterator

In the early versions of the library, we had only channels and we tried to implement various transformations of asynchronous sequences as functions that take one channel as an argument and return another channel as a result. It means that, for example, a

filter

operator would run in its own coroutine.

The performance of such an operator was far from great, especially compared to just writing an

if

statement. In a hindsight, it is not surprising, because a channel is a synchronization primitive. Any channel, even an implementation that is optimized for a single producer and a single consumer, must support concurrent communicating coroutines and a data transfer between them needs synchronization, which is expensive in modern multicore systems. When you start building your application architecture on top of the asynchronous data streams, the need to have transformations naturally appears, and the channel costs start to accrue.

flow ( producer ).transformImperatively( transformer ).collect( ... )