Package hu.akarnokd.kotlin.flow

Types

BehaviorSubject

@FlowPreview()

class BehaviorSubject<T> : AbstractFlow<T> , SubjectAPI<T>

Caches one item and replays it to fresh collectors.

BufferingResumableCollector

open class BufferingResumableCollector<T> constructor(capacity: Int) : Resumable

A collector that can buffer a fixed number of items and drain it with much less need to coordinate the resumption per items.

ConnectableFlow

interface ConnectableFlow<T> : Flow<T>

A possibly multicasting Flow that begins emitting values only when the connect method is called.

GroupedFlow

interface GroupedFlow<K, V> : Flow<V>

Represents a Flow with a group key.

MulticastSubject

@FlowPreview()

class MulticastSubject<T>(expectedCollectors: Int) : AbstractFlow<T> , SubjectAPI<T>

A subject implementation that awaits a certain number of collectors to start consuming, then allows the producer side to deliver items to them.

ParallelFlow

interface ParallelFlow<out T>

Base interface for parallel stages in an operator

PublishSubject

@FlowPreview()

class PublishSubject<T> : AbstractFlow<T> , SubjectAPI<T>

Multicasts items to any number of collectors when they are ready to receive.

ReplaySubject

@FlowPreview()

class ReplaySubject<T> : AbstractFlow<T> , SubjectAPI<T>

Caches and replays some or all items to collectors.

Resumable

open class Resumable : AtomicReference<Continuation<Any>>

Primitive for suspending and resuming coroutines on demand

ResumableCollector

open class ResumableCollector<T> : Resumable

A collector that hosts a signal (value/error/completion) and allows waiting for the signal and a consumer to be ready to receive them.

SubjectAPI

interface SubjectAPI<T> : FlowCollector<T> , Flow<T>

Base interface for suspendable push signals emit, emitError and complete.

UnicastSubject

class UnicastSubject<T> : AbstractFlow<T> , SubjectAPI<T>

Buffers items until a single collector collects them.

UnicastWorkSubject

class UnicastWorkSubject<T> : AbstractFlow<T> , SubjectAPI<T>

Buffers items until a single collector collects them.

Functions

concatArrayEager

@FlowPreview()

fun <T> concatArrayEager(vararg sources: Flow<T>): Flow<T>

Launches all sources at once and emits all items from a source before items of the next are emitted.

concatMap

@ExperimentalCoroutinesApi()

fun <T, R> ParallelFlow<T>.concatMap(mapper: suspend (T) -> Flow<R>): ParallelFlow<R>

Maps the upstream value on each rail onto a Flow and emits their values in order on the same rail.

concatMapEager

@FlowPreview()

fun <T, R> Flow<T>.concatMapEager(mapper: suspend (T) -> Flow<R>): Flow<R>

Maps the upstream values into Flows and launches them all at once, then emits items from a source before items of the next are emitted.

concatWith

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

Emit values from the upstream followed by values from the other flow.

filter

fun <T> ParallelFlow<T>.filter(predicate: suspend (T) -> Boolean): ParallelFlow<T>

Filters the values of the upstream in parallel.

flatMapDrop

@FlowPreview()

fun <T, R> Flow<T>.flatMapDrop(mapper: suspend (T) -> Flow<R>): Flow<R>

Maps items from the upstream to Flow and relays its items while dropping upstream items until the current inner Flow completes.

groupBy

@FlowPreview()

fun <T, K> Flow<T>.groupBy(keySelector: suspend (T) -> K): Flow<GroupedFlow<K, T>>

Groups the upstream values into their own Flows keyed by the value returned by the keySelector function.

@FlowPreview()

fun <T, K, V> Flow<T>.groupBy(keySelector: suspend (T) -> K, valueSelector: suspend (T) -> V): Flow<GroupedFlow<K, V>>

Groups the mapped upstream values into their own Flows keyed by the value returned by the keySelector function.

map

fun <T, R> ParallelFlow<T>.map(mapper: suspend (T) -> R): ParallelFlow<R>

Maps the values of the upstream in parallel.

mergeArray

@FlowPreview()

fun <T> mergeArray(vararg sources: Flow<T>): Flow<T>

Merges multiple sources in an unbounded manner.

multicast

@FlowPreview()

fun <T, R> Flow<T>.multicast(subjectProvider: () -> SubjectAPI<T>, transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

Shares a single collector towards the upstream source and multicasts values to any number of consumers which then can produce the output flow of values.

onBackpressureDrop

@FlowPreview()

fun <T> Flow<T>.onBackpressureDrop(): Flow<T>

Drops items from the upstream when the downstream is not ready to receive them.

parallel

fun <T> Flow<T>.parallel(parallelism: Int, runOn: (Int) -> CoroutineDispatcher): ParallelFlow<T>

Consumes the upstream and dispatches individual items to a parallel rail of the parallel flow for further consumption.

publish

@FlowPreview()

fun <T, R> Flow<T>.publish(transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

Shares a single collector towards the upstream source and multicasts values to any number of consumers which then can produce the output flow of values.

@FlowPreview()

fun <T, R> Flow<T>.publish(expectedCollectors: Int, transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

Shares a single collector towards the upstream source and multicasts values to any number of consumers which then can produce the output flow of values.

range

fun range(start: Int, count: Int): Flow<Int>

Generates a range of ever increasing integer values.

reduce

@FlowPreview()

fun <T> ParallelFlow<T>.reduce(combine: suspend (T, T) -> T): Flow<T>

Reduce the values within the parallel rails and then reduce the rails to a single result value.

fun <T, R> ParallelFlow<T>.reduce(seed: suspend () -> R, combine: suspend (R, T) -> R): ParallelFlow<R>

Reduces the source items into a single value on each rail and emits those.

replay

@FlowPreview()

fun <T, R> Flow<T>.replay(transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

Shares a single collector towards the upstream source and multicasts cached values to any number of consumers which then can produce the output flow of values.

@FlowPreview()

fun <T, R> Flow<T>.replay(maxSize: Int, transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

Shares a single collector towards the upstream source and multicasts up to a given maxSize number of cached values to any number of consumers which then can produce the output flow of values.

@FlowPreview()

fun <T, R> Flow<T>.replay(maxTime: Long, unit: TimeUnit, transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

Shares a single collector towards the upstream source and multicasts up to maxTime old cached values to any number of consumers which then can produce the output flow of values.

@FlowPreview()

fun <T, R> Flow<T>.replay(maxSize: Int, maxTime: Long, unit: TimeUnit, transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

Shares a single collector towards the upstream source and multicasts up to a given maxSize number and up to maxTime old cached values to any number of consumers which then can produce the output flow of values.

@FlowPreview()

fun <T, R> Flow<T>.replay(maxSize: Int, maxTime: Long, unit: TimeUnit, timeSource: (TimeUnit) -> Long, transform: suspend (Flow<T>) -> Flow<R>): Flow<R>

sequential

@FlowPreview()

fun <T> ParallelFlow<T>.sequential(): Flow<T>

Consumes the parallel upstream and turns it into a sequential flow again.

startCollectOn

@FlowPreview()

fun <T> Flow<T>.startCollectOn(dispatcher: CoroutineDispatcher): Flow<T>

Stats collecting the upstream on the specified dispatcher.

takeUntil

@FlowPreview()

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

Consumes the main source until the other source emits an item or completes.

timer

fun timer(timeout: Long, unit: TimeUnit): Flow<Long>

Signal 0L after the given time passed

toList

fun <T> Flow<T>.toList(): Flow<List<T>>

Collects all items of the upstream into a list.

transform

fun <T, R> ParallelFlow<T>.transform(callback: suspend FlowCollector<R>.(T) -> Unit): ParallelFlow<R>

Transform each upstream item into zero or more emits for the downstream in parallel.