-
Notifications
You must be signed in to change notification settings - Fork 1.9k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add chunked and windowed operators #1558
Closed
Closed
Changes from all commits
Commits
Show all changes
8 commits
Select commit
Hold shift + click to select a range
c0bf01b
Merge pull request #2 from Kotlin/develop
circusmagnus 86a503d
Add chunked and windowed operators
circusmagnus dbe84ea
Add changes requested by R Elizarov
circusmagnus 6155c22
Generate api dump
circusmagnus c74bfcf
Use ArrayDeque from stdlib instead of custom RingBuffer
circusmagnus 07f72e9
Change package name to match convention
circusmagnus 9392bbd
Remove RingBuffer from coroutines lib
circusmagnus 8e7ee37
Update coroutines api with chunked and windowed operators
circusmagnus File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
100 changes: 100 additions & 0 deletions
100
kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| @file:JvmMultifileClass | ||
| @file:JvmName("FlowKt") | ||
|
|
||
| package kotlinx.coroutines.flow | ||
|
|
||
| import kotlinx.coroutines.* | ||
| import kotlin.jvm.* | ||
| import kotlin.math.* | ||
|
|
||
| /** | ||
| * Returns a flow of lists each not exceeding the given [size]. | ||
| * The last list in the resulting flow may have less elements than the given [size]. | ||
| * | ||
| * @param size the number of elements to take in each list, must be positive and can be greater than the number of elements in this flow. | ||
| */ | ||
|
|
||
| @FlowPreview | ||
| public fun <T> Flow<T>.chunked(size: Int): Flow<List<T>> = chunked(size) { it.toList() } | ||
|
|
||
| /** | ||
| * Chunks a flow of elements into flow of lists, each not exceeding the given [size] | ||
| * and applies the given [transform] function to an each. | ||
| * | ||
| * Note that the list passed to the [transform] function is ephemeral and is valid only inside that function. | ||
| * You should not store it or allow it to escape in some way, unless you made a snapshot of it. | ||
| * The last list may have less elements than the given [size]. | ||
| * | ||
| * This is more efficient, than using flow.chunked(n).map { ... } | ||
| * | ||
| * @param size the number of elements to take in each list, must be positive and can be greater than the number of elements in this flow. | ||
| */ | ||
|
|
||
| @FlowPreview | ||
| public fun <T, R> Flow<T>.chunked(size: Int, transform: suspend (List<T>) -> R): Flow<R> { | ||
| require(size > 0) { "Size should be greater than 0, but was $size" } | ||
| return windowed(size, size, true, transform) | ||
| } | ||
|
|
||
| /** | ||
| * Returns a flow of snapshots of the window of the given [size] | ||
| * sliding along this flow with the given [step], where each | ||
| * snapshot is a list. | ||
| * | ||
| * Several last lists may have less elements than the given [size]. | ||
| * | ||
| * Both [size] and [step] must be positive and can be greater than the number of elements in this flow. | ||
| * @param size the number of elements to take in each window | ||
| * @param step the number of elements to move the window forward by on an each step | ||
| * @param partialWindows controls whether or not to keep partial windows in the end if any. | ||
| */ | ||
|
|
||
| @FlowPreview | ||
| public fun <T> Flow<T>.windowed(size: Int, step: Int, partialWindows: Boolean): Flow<List<T>> = | ||
| windowed(size, step, partialWindows) { it.toList() } | ||
|
|
||
| /** | ||
| * Returns a flow of results of applying the given [transform] function to | ||
| * an each list representing a view over the window of the given [size] | ||
| * sliding along this collection with the given [step]. | ||
| * | ||
| * Note that the list passed to the [transform] function is ephemeral and is valid only inside that function. | ||
| * You should not store it or allow it to escape in some way, unless you made a snapshot of it. | ||
| * Several last lists may have less elements than the given [size]. | ||
| * | ||
| * This is more efficient, than using flow.windowed(...).map { ... } | ||
| * | ||
| * Both [size] and [step] must be positive and can be greater than the number of elements in this collection. | ||
| * @param size the number of elements to take in each window | ||
| * @param step the number of elements to move the window forward by on an each step. | ||
| * @param partialWindows controls whether or not to keep partial windows in the end if any. | ||
| */ | ||
|
|
||
| @OptIn(ExperimentalStdlibApi::class) | ||
| @FlowPreview | ||
| public fun <T, R> Flow<T>.windowed(size: Int, step: Int, partialWindows: Boolean, transform: suspend (List<T>) -> R): Flow<R> { | ||
| require(size > 0 && step > 0) { "Size and step should be greater than 0, but was size: $size, step: $step" } | ||
|
|
||
| return flow { | ||
| val buffer = ArrayDeque<T>(size) | ||
| val toDrop = min(step, size) | ||
| val toSkip = max(step - size, 0) | ||
| var skipped = toSkip | ||
|
|
||
| collect { value -> | ||
| if (toSkip == skipped) buffer.addLast(value) | ||
| else skipped++ | ||
|
|
||
| if (buffer.size == size) { | ||
| emit(transform(buffer)) | ||
| repeat(toDrop) { buffer.removeFirst() } | ||
| skipped = 0 | ||
| } | ||
| } | ||
|
|
||
| while (partialWindows && buffer.isNotEmpty()) { | ||
| emit(transform(buffer)) | ||
| repeat(min(toDrop, buffer.size)) { buffer.removeFirst() } | ||
| } | ||
| } | ||
| } | ||
59 changes: 59 additions & 0 deletions
59
kotlinx-coroutines-core/common/test/flow/operators/ChunkedTest.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| package kotlinx.coroutines.flow.operators | ||
|
|
||
| import kotlinx.coroutines.* | ||
| import kotlinx.coroutines.channels.Channel | ||
| import kotlinx.coroutines.flow.* | ||
| import kotlin.test.Test | ||
| import kotlin.test.assertEquals | ||
|
|
||
| class ChunkedTest : TestBase() { | ||
|
|
||
| private val flow = flow { | ||
| emit(1) | ||
| emit(2) | ||
| emit(3) | ||
| emit(4) | ||
| } | ||
|
|
||
| @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()) | ||
| } | ||
|
|
||
| @Test | ||
| 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()) | ||
| } | ||
|
|
||
| @Test | ||
| fun testErrorCancelsUpstream() = runTest { | ||
| val latch = Channel<Unit>() | ||
| val flow = flow { | ||
| coroutineScope { | ||
| launch(start = CoroutineStart.ATOMIC) { | ||
| latch.send(Unit) | ||
| hang { expect(3) } | ||
| } | ||
| emit(1) | ||
| expect(1) | ||
| emit(2) | ||
| expectUnreached() | ||
| } | ||
| }.chunked<Int, Int>(2) { chunk -> | ||
| expect(2) // 2 | ||
| latch.receive() | ||
| throw TestException() | ||
| }.catch { emit(42) } | ||
|
|
||
| assertEquals(42, flow.single()) | ||
| finish(4) | ||
| } | ||
| } |
82 changes: 82 additions & 0 deletions
82
kotlinx-coroutines-core/common/test/flow/operators/WindowedTest.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package kotlinx.coroutines.flow.operators | ||
|
|
||
| import kotlinx.coroutines.* | ||
| import kotlinx.coroutines.channels.Channel | ||
| import kotlinx.coroutines.flow.* | ||
| import kotlin.test.Test | ||
| import kotlin.test.assertEquals | ||
|
|
||
| class WindowedTest : TestBase() { | ||
|
|
||
| private val flow = flow { | ||
| emit(1) | ||
| emit(2) | ||
| emit(3) | ||
| emit(4) | ||
| } | ||
|
|
||
| @Test | ||
| 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()) | ||
| } | ||
|
|
||
| @Test | ||
| fun `Emits correct sum with overlapping non partial windows`() = runTest { | ||
| assertEquals(15, flow.windowed(3, 1, false) { window -> | ||
| window.sum() | ||
| }.sum()) | ||
| } | ||
|
|
||
| @Test | ||
| fun `Emits correct sum with overlapping partial windows`() = runTest { | ||
| assertEquals(13, flow.windowed(3, 2, true) { window -> | ||
| window.sum() | ||
| }.sum()) | ||
| } | ||
|
|
||
| @Test | ||
| fun `Emits correct number of overlapping windows for long sequence of overlapping partial windows`() = runTest { | ||
| val elements = generateSequence(1) { it + 1 }.take(100) | ||
| val flow = elements.asFlow().windowed(100, 1, true) | ||
| assertEquals(100, flow.count()) | ||
| } | ||
|
|
||
| @Test | ||
| fun `Emits correct sum with partial windows set apart`() = runTest { | ||
| assertEquals(7, flow.windowed(2, 3, true) { window -> | ||
| window.sum() | ||
| }.sum()) | ||
| } | ||
|
|
||
| @Test | ||
| fun testErrorCancelsUpstream() = runTest { | ||
| val latch = Channel<Unit>() | ||
| val flow = flow { | ||
| coroutineScope { | ||
| launch(start = CoroutineStart.ATOMIC) { | ||
| latch.send(Unit) | ||
| hang { expect(3) } | ||
| } | ||
| emit(1) | ||
| expect(1) | ||
| emit(2) | ||
| expectUnreached() | ||
| } | ||
| }.windowed<Int, Int>(2, 3, false) { window -> | ||
| expect(2) // 2 | ||
| latch.receive() | ||
| throw TestException() | ||
| }.catch { emit(42) } | ||
|
|
||
| assertEquals(42, flow.single()) | ||
| finish(4) | ||
| } | ||
| } |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Quick observation. If the upstream flow throws an exception, you could end up with a partial buffer here.
what are your thoughts of adding this before the collect
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I have fear, that we can loose exception this way or otherwise obscure it.
What if emitting partial buffer gets cancelled on some suspension point, before we rethrow exception? Or even, what if exception will be thrown from emitting our partial buffer, before we manage to rethrow original exception? Perhaps we could fix this by using a finally block somewhere. But, I think, it would be simpler, to just let exception propagate right away, and loose unfortunate partial buffer.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Loosing partial buffer is somewhat analogous to coroutineScope cancelling all its children immediately, on detecting exception anywhere. It is just more consistent this way.