Flow.chunked operator with size limit

[azulkarnyaev]

It would be useful to have an optional transformation in Flow.buffer method to aggregate the buffered items like in kotlin.sequences.Sequence.chunked.
I mean,
fun buffer(capacity: Int = BUFFERED, suspend transform: (List<T>) -> R) : Flow<T>

Then we can write

runBlocking {
        (1..100).asFlow().buffer(capacity = 10) { it.sum() }.collect { println(it) }
    }

with result 55, 155, 255, ... , 955

[elizarov]

That would be a separate operator that we'll call chunked or something like that (as we generally follow stdlib naming convention). This operator will be totally unrelated to the buffer operator. Unlike buffer, this operator will be fully sequential.

P.S. Rx has a whole set of bufferXxx operators that actually correspond to chunked/windowed in Kotlin. On the other hand, buffer/conflate operators in Kotlin flows somewhat correspond to Rx onBackpressureBuffer operators.

[elizarov]

Also, I've forgot to ask what be your use-case for such an operator?

[azulkarnyaev]

Thank you very much for the response!
About the use case: I need to handle stream of vectors (csv row) from a tcp socket and write aggregated statistics for every chunk of n messages to files.

[elizarov]

Are you sure you need Flow for this? Wouldn't a Sequence from Kotlin standard library work for you?

[azulkarnyaev]

Well, I'm using Ktor as a server for socket connection and using coroutines to write data to a file. Pseudo code for my task:

launch {
    val socket = server.accept()
    val input = socket.openReadChannel()
    flow {
          while (true) {
                val line = input.readUTF8Line()
                emit(line)
           }
    }.map { 
          convertToDomainObject(it) 
    }. chunked(1000) {
          aggregateToDomainObjects(it)
    }.collect {
          writeToFile(it)
    }
}

Yes, I can use just sync Sequence. But then I need to provide a back pressure mechanism manually: what if I receive messages faster than I store them? Hopefully, chunked() method will be the same as buffer() and provide a way for back pressure out of the box.

[elizarov]

@azulkarnyaev Thanks for explanation. It does make sense.

[circusmagnus]

I would second this issue - I have a use case, where I'm receiving subsequent snapshots of a database and I need to produce classes representing diffs between those snapshots.

So I need to cache two subsequent emissions, emit them as Pair(first, second), remove the first emission, and wait for the third emission to emit another Pair(second, third). With collections, I do get it with windowed function.

I do not control frequency of emissions and they must happen on a background thread -> hence Flow is needed.

I think it could be generalized into something like this:

fun <T> Flow<T>.windowed(size: Int, step: Int): Flow<List<T>> = flow {
    // check that size and step are > 0
    val queue = ArrayDeque<T>(size)
    val toSkip = max(step - size, 0) < if sbd would like to skip some elements before getting another window, by serving step greater than size, then why not?
    val toRemove = min(step, size)
    var skipped = 0
    collect { element ->
        if(queue.size < size && skipped == toSkip) {
            queue.add(element)
        }
        else if (queue.size < size && skipped < toSkip) {
        skipped++
    }

        if(queue.size == size) {
            emit(queue.toList())
            repeat(toRemove) { queue.remove() }
            skipped = 0
        }
    }
} 

Intended use:

 flow.windowed(size = 2, step = 1)
    .map { listOfTwoNeighboringEmissions ->
        computeDiff()
} 

[zach-klippenstein]

@circusmagnus That sounds like a use case for scan more than windowing.

[circusmagnus]

Almost. Scan requires me to either provide an initial value, which I do not have (an empty diff is stupid, I just need to swallow first dB emission and wait for the second one to produce a diff) or to emit the same type, as I am receiving (scanReduce), which is also a no go, as I get a list of entities, but want to emit changes between them.

[tunjid]

Would this work?

            flow
                .scan(listOf<Item>()) { oldItems, newItem ->
                    if (oldItems.size >= BUFFER_COUNT) listOf(newItem)
                    else oldItems + newItem
                }
                .filter { it.size ==  BUFFER_COUNT}