is there a flow equivalent to <http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/core/Obse...

# coroutines

[octylFractal]

is there a flow equivalent to http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/core/Observable.html#buffer-long-java.util.concurrent.TimeUnit-int- ?

[gildor]

Buffer with timespan? I don’t think so

[octylFractal]

I just wrote https://gist.github.com/octylFractal/e95df0727354b9d6b76f2ede3cae8b89, but it's probably got some bugs 😛

[octylFractal]

not exactly the same behavior, but it's what I was looking for

[Dominaezzz]

@ExperimentalCoroutinesApi
@ExperimentalTime
fun <T> Flow<T>.bufferChunks(count: Int, maxDuration: Duration): Flow<List<T>> {
	return flow {
		coroutineScope {
			val channel = produceIn(this)
			while (!channel.isClosedForReceive) {
				val chunk = mutableListOf<T>()
				withTimeout(maxDuration) {
					for (item in channel) {
						chunk.add(item)
						if (chunk.size >= count) break
					}
				}
				emit(chunk)
			}
		}
	}
}

[Dominaezzz]

I got bored and re-implemented. 🙂

[Dominaezzz]

A decent name for this operator would be nagle I think.

[octylFractal]

ah, that is much simpler.

[Circusmagnus]

I am working on PR to add chunk operator to Flow with time, size and natural batching limit FYI. Comments are welcome: https://github.com/Kotlin/kotlinx.coroutines/issues/1302