kitecore/io.github.yuroyami.kitecore.flow

Package-level declarations

Functions

asResult

fun <T> Flow<T>.asResult(): Flow<Result<T>>

Wraps each upstream value in Result.success and a terminal upstream failure in a single Result.failure.

chunked

fun <T> Flow<T>.chunked(size: Int): Flow<List<T>>

Groups values into lists of size elements, emitting the final partial list on completion.

fun <T> Flow<T>.chunked(window: Duration, maxSize: Int = Int.MAX_VALUE): Flow<List<T>>

Collects values into lists bounded by a time window and an optional maxSize.

collectIn

fun <T> Flow<T>.collectIn(scope: CoroutineScope, action: suspend (T) -> Unit): Job

Launches collection of the flow in scope, invoking action for each value, and returns the Job.

combine6

fun <T1, T2, T3, T4, T5, T6, R> combine6(flow: Flow<T1>, flow2: Flow<T2>, flow3: Flow<T3>, flow4: Flow<T4>, flow5: Flow<T5>, flow6: Flow<T6>, transform: suspend (T1, T2, T3, T4, T5, T6) -> R): Flow<R>

Combines the latest values of six flows through transform.

combine7

fun <T1, T2, T3, T4, T5, T6, T7, R> combine7(flow: Flow<T1>, flow2: Flow<T2>, flow3: Flow<T3>, flow4: Flow<T4>, flow5: Flow<T5>, flow6: Flow<T6>, flow7: Flow<T7>, transform: suspend (T1, T2, T3, T4, T5, T6, T7) -> R): Flow<R>

Combines the latest values of seven flows through transform.

combine8

fun <T1, T2, T3, T4, T5, T6, T7, T8, R> combine8(flow: Flow<T1>, flow2: Flow<T2>, flow3: Flow<T3>, flow4: Flow<T4>, flow5: Flow<T5>, flow6: Flow<T6>, flow7: Flow<T7>, flow8: Flow<T8>, transform: suspend (T1, T2, T3, T4, T5, T6, T7, T8) -> R): Flow<R>

Combines the latest values of eight flows through transform.

combine9

fun <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> combine9(flow: Flow<T1>, flow2: Flow<T2>, flow3: Flow<T3>, flow4: Flow<T4>, flow5: Flow<T5>, flow6: Flow<T6>, flow7: Flow<T7>, flow8: Flow<T8>, flow9: Flow<T9>, transform: suspend (T1, T2, T3, T4, T5, T6, T7, T8, T9) -> R): Flow<R>

Combines the latest values of nine flows through transform.

combineStates

fun <A, B, R> combineStates(a: StateFlow<A>, b: StateFlow, transform: (A, B) -> R): StateFlow<R>

Returns a read-only StateFlow whose value combines a and b through transform.

fun <A, B, C, R> combineStates(a: StateFlow<A>, b: StateFlow, c: StateFlow<C>, transform: (A, B, C) -> R): StateFlow<R>

Returns a read-only StateFlow whose value combines a, b, and c through transform.

fun <A, B, C, D, R> combineStates(a: StateFlow<A>, b: StateFlow, c: StateFlow<C>, d: StateFlow<D>, transform: (A, B, C, D) -> R): StateFlow<R>

Returns a read-only StateFlow whose value combines four sources through transform.

concatWith

fun <T> Flow<T>.concatWith(other: Flow<T>): Flow<T>

Emits the upstream values followed by the values of other.

fun <T> Flow<T>.concatWith(vararg others: Flow<T>): Flow<T>

Emits the upstream values followed by the values of each flow in others, in order.

countdownFlow

fun countdownFlow(from: Int, interval: Duration): Flow<Int>

Returns a flow that counts down from from to 0, one number every interval.

decrement

fun MutableStateFlow<Int>.decrement(by: Int = 1): Int

fun MutableStateFlow<Long>.decrement(by: Long = 1): Long

Subtracts by from the value and returns the new value.

deferFlow

fun <T> deferFlow(provider: suspend () -> Flow<T>): Flow<T>

Returns a cold flow that obtains the actual flow from provider at collection time.

dropFor

fun <T> Flow<T>.dropFor(duration: Duration): Flow<T>

Drops upstream values for duration, then mirrors the upstream.

dropUntil

fun <T> Flow<T>.dropUntil(signal: Flow<*>): Flow<T>

Drops upstream values until signal emits its first value, then mirrors the upstream.

filterFailure

fun <T> Flow<Result<T>>.filterFailure(): Flow<Throwable>

Emits the exception of every failed Result and drops successes.

filterIndexed

fun <T> Flow<T>.filterIndexed(predicate: suspend (index: Int, value: T) -> Boolean): Flow<T>

Emits only the values for which predicate returns true, passing each value's 0-based position.

filterSuccess

fun <T> Flow<Result<T>>.filterSuccess(): Flow<T>

Emits the value of every successful Result and drops failures.

firstOrDefault

suspend fun <T> Flow<T>.firstOrDefault(default: T): T

Returns the first value of the flow, or default if the flow completes without emitting.

flattenIterable

fun <T> Flow<Iterable<T>>.flattenIterable(): Flow<T>

Emits every element of every upstream iterable, in order.

flowFromSuspend

fun <T> flowFromSuspend(block: suspend () -> T): Flow<T>

Returns a cold flow that invokes block and emits its single result.

getAndSet

fun <T> MutableStateFlow<T>.getAndSet(value: T): T

Sets the value to value and returns the previous value.

increment

fun MutableStateFlow<Int>.increment(by: Int = 1): Int

fun MutableStateFlow<Long>.increment(by: Long = 1): Long

Adds by to the value and returns the new value.

intersperse

fun <T> Flow<T>.intersperse(separator: T): Flow<T>

Emits separator between each pair of consecutive upstream values.

intervalFlow

fun intervalFlow(period: Duration, initialDelay: Duration = Duration.ZERO): Flow<Long>

Returns an infinite flow that emits sequential numbers starting at 0, one every period.

lastOrDefault

suspend fun <T> Flow<T>.lastOrDefault(default: T): T

Returns the last value of the flow, or default if the flow completes without emitting.

mapAsync

fun <T, R> Flow<T>.mapAsync(concurrency: Int, transform: suspend (T) -> R): Flow<R>

Maps values through transform with up to concurrency invocations in flight, preserving upstream order.

mapAsyncUnordered

fun <T, R> Flow<T>.mapAsyncUnordered(concurrency: Int, transform: suspend (T) -> R): Flow<R>

Maps values through transform with up to concurrency invocations in flight, emitting results in completion order.

mapIndexed

fun <T, R> Flow<T>.mapIndexed(transform: suspend (index: Int, value: T) -> R): Flow<R>

Maps each value through transform together with its 0-based position in the flow.

mapState

fun <T, R> StateFlow<T>.mapState(transform: (T) -> R): StateFlow<R>

Returns a read-only StateFlow view whose value is transform applied to this flow's value.

mapToUnit

fun <T> Flow<T>.mapToUnit(): Flow<Unit>

Replaces every upstream value with Unit.

neverFlow

fun neverFlow(): Flow<Nothing>

Returns a flow that never emits and never completes.

onEachIndexed

fun <T> Flow<T>.onEachIndexed(action: suspend (index: Int, value: T) -> Unit): Flow<T>

Invokes action with each value and its 0-based position, then emits the value unchanged.

onFirst

fun <T> Flow<T>.onFirst(action: suspend (T) -> Unit): Flow<T>

Invokes action with the first value only, then emits all values unchanged.

pairwise

fun <T> Flow<T>.pairwise(): Flow<Pair<T, T>>

Emits consecutive pairs of upstream values, starting with the second value.

fun <T, R> Flow<T>.pairwise(transform: suspend (previous: T, current: T) -> R): Flow<R>

Applies transform to each pair of consecutive upstream values.

retryWithBackoff

fun <T> Flow<T>.retryWithBackoff(retries: Int, initialDelay: Duration, factor: Double = 2.0, maxDelay: Duration = Duration.INFINITE, retryIf: (Throwable) -> Boolean = { true }): Flow<T>

Retries a failed upstream up to retries times with exponentially growing delays.

runningCount

fun <T> Flow<T>.runningCount(): Flow<Int>

Emits the running number of values collected so far, starting at 1 for the first value.

startWith

fun <T> Flow<T>.startWith(value: T): Flow<T>

Emits value before the upstream values.

fun <T> Flow<T>.startWith(vararg values: T): Flow<T>

Emits values in order before the upstream values.

stateFlowOf

fun <T> stateFlowOf(value: T): StateFlow<T>

Returns a StateFlow that always holds value.

takeFor

fun <T> Flow<T>.takeFor(duration: Duration): Flow<T>

Mirrors the upstream for duration, then completes.

takeUntil

fun <T> Flow<T>.takeUntil(signal: Flow<*>): Flow<T>

Mirrors the upstream until signal emits its first value, then completes.

takeWhileInclusive

fun <T> Flow<T>.takeWhileInclusive(predicate: suspend (T) -> Boolean): Flow<T>

Emits values while predicate returns true, including the first value for which it returns false.

throttleFirst

fun <T> Flow<T>.throttleFirst(window: Duration): Flow<T>

Emits the first value, then drops every value that arrives within window of the last emission.

throttleLatest

fun <T> Flow<T>.throttleLatest(window: Duration): Flow<T>

Emits the most recent value once per window, dropping intermediate values.

timeoutBetweenEmissions

fun <T> Flow<T>.timeoutBetweenEmissions(timeout: Duration): Flow<T>

Fails with kotlinx.coroutines.TimeoutCancellationException when the gap between upstream emissions exceeds timeout.

timerFlow

fun timerFlow(delay: Duration): Flow<Unit>

Returns a flow that emits a single Unit after delay and then completes.

toggle

fun MutableStateFlow<Boolean>.toggle(): Boolean

Flips the boolean value and returns the new value.

windowed

fun <T> Flow<T>.windowed(size: Int, step: Int = 1): Flow<List<T>>

Emits sliding windows of size elements, advancing by step elements between windows.

withLatestFrom

fun <T, U> Flow<T>.withLatestFrom(other: Flow): Flow<Pair<T, U>>

Pairs each upstream value with the latest value of other.

fun <T, U, R> Flow<T>.withLatestFrom(other: Flow, transform: suspend (T, U) -> R): Flow<R>

Combines each upstream value with the latest value of other through transform.

withPrevious

fun <T> Flow<T>.withPrevious(): Flow<Pair<T?, T>>

Emits each value together with the value that preceded it, null for the first value.