kmath-coroutines/space.kscience.kmath.coroutines/AsyncFlow

AsyncFlow

class AsyncFlow<out T> : Flow<T> (source)

Functions

inline fun <T, R> Flow<T>.async(dispatcher: CoroutineDispatcher = Dispatchers.Default, crossinline block: suspend CoroutineScope.(T) -> R): AsyncFlow<R>

fun Flow<Double>.chunked(bufferSize: Int): Flow<DoubleBuffer>

Specialized flow chunker for real buffer

fun <T> Flow<T>.chunked(bufferSize: Int, bufferFactory: BufferFactory<T>): Flow<Buffer<T>>

Collect incoming flow into fixed size chunks

open suspend override fun collect(collector: FlowCollector<T>)

inline suspend fun <T> AsyncFlow<T>.collect(concurrency: Int, crossinline action: suspend (value: T) -> Unit)

suspend fun <T> AsyncFlow<T>.collect(concurrency: Int, collector: FlowCollector<T>)

fun <T> Flow<T>.cumulativeSum(group: GroupOps<T>): Flow<T>

inline fun <T, R> AsyncFlow<T>.map(crossinline action: (T) -> R): AsyncFlow<R>

inline fun <T, R> Flow<T>.mapParallel(dispatcher: CoroutineDispatcher = Dispatchers.Default, crossinline transform: suspend (T) -> R): Flow<R>

@ExperimentalCoroutinesApi

fun <T, S : Ring<T>, ScaleOperations<T>> Flow<T>.mean(space: S): Flow<T>

fun <T> Flow<Buffer<T>>.spread(): Flow<T>

Flat map a Flow of Buffer into continuous Flow of elements

fun <T> Flow<T>.windowed(window: Int): Flow<Buffer<T>>

Map a flow to a moving window buffer. The window step is one. To get different steps, one could use skip operation.