hoc081098 / hoc081098/FlowExt

Add time windowed chunking, `groupedWithin`

Open
#185 2 comments 0 reactions 0 assignees View on GitHub
enhancement PRs-are-welcome
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.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.