Add time windowed chunking, `groupedWithin`
- Dominant language
- Kotlin
- Stars
- 419
- Forks
- 27
- Avg merge
- 17m
- Merged PRs (30d)
- 1
Description
Similar to `groupedWithin` in Akka, I think this is a very useful utility to have.
[Akka Docs](https://doc.akka.io/api/akka/2.8/akka/stream/scaladsl/Source.html#groupedWithin(n:Int,d:scala.concurrent.duration.FiniteDuration):FlowOps.this.Repr%5BSeq%5BOut%5D%5D)
**Proposed solution**
```kotlin
fun Flow.groupedWithin(size: Int, limit: Duration): Flow> { ... }
```
implementation can be based on https://github.com/Kotlin/kotlinx.coroutines/issues/1290#issuecomment-1309290613
I have modified that code slightly, I can help work on a solution based on a channel flow.
**Behavior**
- Once flow reaches **size** items, it emits.
- If flow can't reach **size** items within **limit** time, it emits the items collected until now, unless there is none.
**Why**
This is very useful when we are bridging the gap between the regular APIs and streaming APIs. For example, assume you have an API to fetch SQS messages, traditionally you would implement it as
```kotlin
suspend fun main() {
val sqsClient = getSqsClient()
while (true) {
val items = sqsClient.poll(10)
process(items)
delay(10.seconds)
}
}
```
instead, we can use
```kotlin
suspend fun main() {
val sqsClient = getSqsClient()
flow {
while (true) {
val items = sqsClient.poll(10)
items.forEach { emit(it) }
delay(10.seconds)
}
}.groupedWithin(128, 30.seconds) {
process(it)
}
}
```
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.