kmath-coroutines/space.kscience.kmath.chains/BufferChain

BufferChain

interface BufferChain<out T> : Chain<T> (source)

Inheritors

BlockingBufferChain

Functions

asSequence

fun <R> Chain<R>.asSequence(): Sequence<R>

Represent a chain as a sequence

async

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

chunked

fun Flow<Float64>.chunked(bufferSize: Int): Flow<Float64Buffer>

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

collect

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

combine

@UnstableKMathAPI

fun <T, R> Chain<T>.combine(mapper: suspend (Chain<T>) -> R): Chain<R>

Map the whole chain

combineWithState

@UnstableKMathAPI

fun <T, S, R> Chain<T>.combineWithState(state: S, stateFork: (S) -> S, mapper: suspend S.(Chain<T>) -> R): Chain<R>

cumulativeSum

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

Return a Flow of a cumulative sum of elements in the flow. The operation is intermediate and stateful.

discard

inline suspend fun <T> Chain<T>.discard(number: Int): Chain<T>

Discard a fixed number of samples

filter

fun <T> Chain<T>.filter(block: (T) -> Boolean): Chain<T>

block must be a pure function or at least not use external random variables, otherwise fork could be broken

fork

abstract suspend override fun fork(): BufferChain<T>

Create a copy of the current chain state. Consuming the resulting chain does not affect the initial chain.

iterator

operator fun <R> Chain<R>.iterator(): Iterator<R>

Represent a chain as regular iterator (uses blocking calls)

map

fun <T, R> Chain<T>.map(func: suspend (T) -> R): Chain<R>

Map the chain result using suspended transformation. Initial chain result can no longer be safely consumed since mapped chain consumes tokens. Accepts regular transformation function.

mapParallel

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

mean

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

Return a Flow of mean values of elements in the flow. The operation is intermediate and stateful.

abstract suspend fun next(): T

Generate next value, changing state if needed

nextBuffer

abstract suspend fun nextBuffer(size: Int): Buffer<T>

nextBuffer

inline suspend fun <T> Chain<T>.nextBuffer(size: Int): Buffer<T>

spread

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

Flat map a Flow of Buffer into continuous Flow of elements

windowed

fun <T> Flow<T>.windowed(window: Int, algebra: Group<T>): 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.

zip

fun <T, U, R> Chain<T>.zip(other: Chain<U>, block: suspend (T, U) -> R): Chain<R>

Zip two chains together using given transformation