core/com.river.core/split

split

fun <T> Flow<T>.split(strategy: GroupStrategy): Flow<Flow<T>>

The split function is used to group the elements emitted by the current Flow into smaller Flows based on the provided GroupStrategy. This function is useful when smaller Flows are preferred over the original one, some scenarios include file processing, database operations, and so on.

The smaller streams are emitted as Flow of elements, with each group containing a maximum number of elements or a time window, depending on the provided GroupStrategy implementation.

The available strategies include:

GroupStrategy.Count: Groups the input flow based on the number of items in each chunk.
GroupStrategy.TimeWindow: Groups the input flow based on a specified time duration.

Note that the streams may not be of exactly the same size, depending on the number of elements emitted and the specified strategy. However, the order of elements in the output flows is preserved from the input flow.

Return

A Flow of Flows, where each one represents the smaller Flows of elements from the original flow.

Example usage:

val flow: Flow<Int> = flowOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val chunkSize = 4

flow.split(GroupStrategy.Count(chunkSize))
    .collectIndexed { index, smallerFlow ->
        print("Flow #${index + 1}: ")

        smallerFlow
            .intersperse(", ")
            .collect(::print)

        println()
    }

 // Output:
 // Flow #1: 1, 2, 3, 4
 // Flow #2: 5, 6, 7, 8
 // Flow #3: 9, 10

In this example, the input flow contains 10 elements. The split function is called with a size of 4. This means that the output flow will emit the Flows, each one containing at most 4 elements.

Parameters

strategy

The GroupStrategy implementation to be used for grouping the elements.

fun <T> Flow<T>.split(size: Int): Flow<Flow<T>>

The split function is used to group the elements emitted by the current Flow into smaller Flows based a max number of items. This function is useful when smaller Flows are preferred over the original one, some scenarios include file processing, database operations, and so on.

The smaller streams are emitted as Flow of elements, with each group containing a maximum number of elements.

Note that the streams may not be of exactly the same size, depending on the number of elements emitted. However, the order of elements in the output flows is preserved from the input flow.

Return

A Flow of Flows, where each one represents the smaller Flows of elements from the original flow.

Example usage:

val flow: Flow<Int> = flowOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val splitSize = 4

flow.split(splitSize)
    .collectIndexed { index, smallerFlow ->
        print("Flow #${index + 1}: ")

        smallerFlow
            .intersperse(", ")
            .collect(::print)

        println()
    }

 // Output:
 // Flow #1: 1, 2, 3, 4
 // Flow #2: 5, 6, 7, 8
 // Flow #3: 9, 10

In this example, the input flow contains 10 elements. The split function is called with a size of 4. This means that the output flow will emit the Flows, each one containing at most 4 elements.

Parameters

size

The maximum number of elements in each Flow.

fun <T> Flow<T>.split(size: Int, duration: Duration): Flow<Flow<T>>

The split function is used to group the elements emitted by the current Flow into smaller Flows based on size and timeout duration. This function is useful when smaller Flows are preferred over the original one, some scenarios include file processing, database operations, and so on.

The smaller streams are emitted as Flow of elements, with each group containing a maximum number of elements or a time window.

Note that the streams may not be of exactly the same size, depending on the number of elements emitted, the timeout, and the specified strategy. However, the order of elements in the output flows is preserved from the input flow.

Return

A Flow of Flows, where each one represents the smaller Flows of elements from the original flow.

Example usage:

val flow: Flow<Int> = flowOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val splitSize = 3

flow.split(splitSize, 100.milliseconds)
    .collectIndexed { index, smallerFlow ->
        print("Flow #${index + 1}: ")

        smallerFlow
            .intersperse(", ")
            .collect(::print)

        println()
    }

In this example, the input flow contains 10 elements. The split function is called with a size of 3 and a duration of 100 milliseconds. This means that the output flow will emit the Flows containing up to 3 elements, or lasting up to 100 milliseconds, whichever comes first.

Parameters

size

The maximum number of elements in each Flow.

duration

The maximum duration of each smaller Flow.