Check args validity right away. Pass ephemeral list to transformer.

Author: circusmagnus

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

@@ -4,18 +4,28 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.flow
 
+fun <T> Flow<T>.chunked(size: Int): Flow<List<T>> = chunked(size) { it.toList() }
 
-fun <T, R> Flow<T>.chunked(size: Int, transform: suspend (List<T>) -> R): Flow<R> = nonOverlappingWindowed(size, size, true, transform)
+fun <T, R> Flow<T>.chunked(size: Int, transform: suspend (List<T>) -> R): Flow<R> =
+        nonOverlappingWindowed(size, size, true, transform)
+
+fun <T> Flow<T>.windowed(size: Int, step: Int, partialWindows: Boolean): Flow<List<T>> =
+        windowed(size, step, partialWindows) { it.toList() }
 
 fun <T, R> Flow<T>.windowed(size: Int, step: Int, partialWindows: Boolean, transform: suspend (List<T>) -> R): Flow<R> =
         if (size <= step) nonOverlappingWindowed(size, step, partialWindows, transform)
         else overlappingWindowed(size, step, partialWindows, transform)
 
 
-private fun <T, R> Flow<T>.nonOverlappingWindowed(size: Int, step: Int, partialWindows: Boolean, transform: suspend (List<T>) -> R): Flow<R> {
+private fun <T, R> Flow<T>.nonOverlappingWindowed(
+        size: Int,
+        step: Int,
+        partialWindows: Boolean,
+        transform: suspend (List<T>) -> R
+): Flow<R> {
+    require(size in 1..step) { "Size should be non-negative, and equal to or lesser than step, but was size: $size, step: $step"}
 
     return flow {
-        require(size in 1..step)
         val window = ArrayList<T>(size)
         val toSkip = step - size
         var skipped = toSkip
@@ -26,39 +36,41 @@ private fun <T, R> Flow<T>.nonOverlappingWindowed(size: Int, step: Int, partialW
             } else skipped++
 
             if (window.size == size) {
-                emit(transform(window.toList()))
+                emit(transform(window))
                 window.clear()
                 skipped = 0
             }
         }
 
         if (partialWindows && window.isNotEmpty()) {
-            emit(transform(window.toList()))
+            emit(transform(window))
         }
     }
 }
 
-private fun <T, R> Flow<T>.overlappingWindowed(size: Int, step: Int, partialWindows: Boolean, transform: suspend (List<T>) -> R) = flow {
-    require(step in 1 until size)
-    val buffer = ArrayList<T>(size)
+private inline fun <T, R> Flow<T>.overlappingWindowed(
+        size: Int,
+        step: Int,
+        partialWindows: Boolean,
+        crossinline transform: suspend (List<T>) -> R
+): Flow<R> {
+    require(step in 1 until size) { "Size should be non-negative, and greater than step, but was size: $size, step: $step"}
+
+    return flow {
+        val buffer = ArrayList<T>(size)
 
-    collect { value ->
-        buffer.add(value)
+        collect { value ->
+            buffer.add(value)
 
-        if (buffer.size == size) {
-            val window = makeWindow(step, buffer)
-            emit(transform(window))
+            if (buffer.size == size) {
+                emit(transform(buffer))
+                repeat(step) { buffer.removeAt(0) }
+            }
         }
-    }
 
-    while (partialWindows && buffer.isNotEmpty()) {
-        val window = makeWindow(step, buffer)
-        emit(transform(window))
+        while (partialWindows && buffer.isNotEmpty()) {
+            emit(transform(buffer))
+            repeat(step) { buffer.removeAt(0) }
+        }
     }
-}
-
-private fun <T> makeWindow(step: Int, buffer: MutableList<T>): List<T> {
-    val window = ArrayList(buffer)
-    repeat(step) { buffer.removeAt(0) }
-    return window
 }

kotlinx-coroutines-core/common/test/flow/operators/ChunkedTest.kt

@@ -5,8 +5,6 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.*
 import kotlin.test.Test
 import kotlin.test.assertEquals
-import kotlin.test.assertNull
-import kotlin.test.assertTrue
 
 class ChunkedTest : TestBase() {
 
@@ -19,20 +17,20 @@ class ChunkedTest : TestBase() {
 
     @Test
     fun `Chunks correct number of emissions with possible partial window at the end`() = runTest {
-        assertEquals(2, flow.chunked(2) { }.count())
-        assertEquals(2, flow.chunked(3) { }.count())
-        assertEquals(1, flow.chunked(5) { }.count())
+        assertEquals(2, flow.chunked(2).count())
+        assertEquals(2, flow.chunked(3).count())
+        assertEquals(1, flow.chunked(5).count())
     }
 
     @Test
-    fun `Throws IllegalArgumentException for chunk of size less than 1`() = runTest {
-        assertFailsWith<IllegalArgumentException>(flow.chunked(0) {})
-        assertFailsWith<IllegalArgumentException>(flow.chunked(-1) {})
+    fun `Throws IllegalArgumentException for chunk of size less than 1`() {
+        assertFailsWith<IllegalArgumentException> { flow.chunked(0) }
+        assertFailsWith<IllegalArgumentException> { flow.chunked(-1) }
     }
 
     @Test
     fun `No emissions with empty flow`() = runTest {
-        assertEquals(0, flowOf<Int>().chunked(2) {}.count())
+        assertEquals(0, flowOf<Int>().chunked(2).count())
     }
 
     @Test

kotlinx-coroutines-core/common/test/flow/operators/WindowedTest.kt

@@ -16,16 +16,16 @@ class WindowedTest : TestBase() {
     }
 
     @Test
-    fun `Throws IllegalArgumentException for window of size or step less than 1`() = runTest {
-        assertFailsWith<IllegalArgumentException>(flow.windowed(0, 1, false) {})
-        assertFailsWith<IllegalArgumentException>(flow.windowed(-1, 2, false) {})
-        assertFailsWith<IllegalArgumentException>(flow.windowed(2, 0, false) {})
-        assertFailsWith<IllegalArgumentException>(flow.windowed(5, -2, false) {})
+    fun `Throws IllegalArgumentException for window of size or step less than 1`() {
+        assertFailsWith<IllegalArgumentException> { flow.windowed(0, 1, false) }
+        assertFailsWith<IllegalArgumentException> { flow.windowed(-1, 2, false) }
+        assertFailsWith<IllegalArgumentException> { flow.windowed(2, 0, false) }
+        assertFailsWith<IllegalArgumentException> { flow.windowed(5, -2, false) }
     }
 
     @Test
     fun `No emissions with empty flow`() = runTest {
-        assertEquals(0, flowOf<Int>().windowed(2, 2, false) {}.count())
+        assertEquals(0, flowOf<Int>().windowed(2, 2, false).count())
     }
 
     @Test