datasourcex-util/com.caplin.integration.datasourcex.util.flow/MutableCompletingSharedFlow

MutableCompletingSharedFlow

class MutableCompletingSharedFlow<T> : CompletingSharedFlow<T> , FlowCollector<T> (source)

CompletingSharedFlow equivalent of MutableSharedFlow, providing emit and complete functionality.

Constructors

constructor(replay: Int = 0, extraBufferCapacity: Int = 0, onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND)

Properties

subscriptionCount

val subscriptionCount: StateFlow<Int>

Functions

bufferingDebounce

fun <T> Flow<T>.bufferingDebounce(timeout: Duration): Flow<List<T>>

Buffers all elements emitted until there is a period of no emissions greater than timeout, then emits all buffered elements within a List.

fun <T> Flow<T>.bufferingDebounce(timeoutMillis: Long): Flow<List<T>>

Buffers all elements emitted until there is a period of no emissions greater than timeoutMillis, then emits all buffered elements within a List.

cast

fun <R> Flow<*>.cast(): Flow<R>

Casts the elements of the current Flow to the specified type R. The elements that cannot be cast to R will cause a ClassCastException.

collect

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

collectSafely

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

complete

suspend fun complete(throwable: Throwable? = null)

conflateKeys

fun <K : Any, V : Any> Flow<MapEvent<K, V>>.conflateKeys(): Flow<MapEvent<K, V>>

Conflates each key in the map individually. This buffers emissions, but due to conflation, the maximum number of queued emissions is equal to the number of keys in the map.

dematerialize

fun <T> Flow<ValueOrCompletion<T>>.dematerialize(): Flow<T>

Dematerializes ValueOrCompletions back into their counterpart Flow events.

dematerializeUnboxed

@DelicateCoroutinesApi

fun <T> Flow<Any?>.dematerializeUnboxed(): Flow<T>

Specialised version of dematerialize which expects unboxed updates. Type information is lost with materializeUnboxed, so must know what type to cast updates to.

demultiplexBy

fun <T, K, R> Flow<T>.demultiplexBy(keySelector: (T) -> K?, flowProducer: suspend FlowCollector<R>.(K, Flow<T>) -> Unit): Flow<R>

Demultiplexes the elements of the flow based on a key selector function. Each element in the flow is assigned a key based on the result of the key selector function. Elements with the same key are grouped together and then processed by the flow producer function.

emit

open suspend override fun emit(value: T)

flatMapFirst

fun <T, R> Flow<T>.flatMapFirst(block: suspend (first: T) -> Flow<R>): Flow<R>

Subscribes to the upstream and when it receives the first element calls block and then emits the returned Flow

fun <T, R> Flow<T>.flatMapFirst(block: suspend (first: T, upstream: Flow<T>) -> Flow<R>): Flow<R>

Applies the given suspend lambda function to the first emitted value of the original flow and the entire original flow, producing a new flow of type R.

flatMapLatestAndMerge

fun <V : Any, R> Flow<SetEvent<V>>.flatMapLatestAndMerge(entryEventTransformer: (SetEvent.EntryEvent<V>) -> Flow<R>): Flow<R>

@JvmName(name = "flatMapLatestAndMergeSet")

fun <V : Any, R> Flow<Set<V>>.flatMapLatestAndMerge(entryEventTransformer: (SetEvent.EntryEvent<V>) -> Flow<R>): Flow<R>

materialize

fun <T> Flow<T>.materialize(): Flow<ValueOrCompletion<T>>

Materialize Flow events into ValueOrCompletions that can be sent over the wire, for example.

materializeUnboxed

@DelicateCoroutinesApi

fun Flow<*>.materializeUnboxed(): Flow<Any?>

Specialised version of materialize which does not box updates. Loses type information so corresponding dematerializeUnboxed must know what type to cast updates to.

onSubscription

fun <T> CompletingSharedFlow<T>.onSubscription(action: suspend FlowCollector<T>.() -> Unit): Flow<T>

retryWithExponentialBackoff

fun <T> Flow<T>.retryWithExponentialBackoff(minMillis: Long = 100, maxMillis: Long = 60000, onRetry: suspend (Throwable, Long) -> Boolean = { _, _ -> true }): Flow<T>

On an error, Retry on the upstream with an exponential delay between each attempt. Behaviour can be modified by proving onRetry - if a call to this returns false it will stop retrying and propagate the error downstream.

runningFoldToMap

@JvmName(name = "runningFoldToMapEntryEvent")

fun <K : Any, V : Any> Flow<MapEvent.EntryEvent<K, V>>.runningFoldToMap(): Flow<Map<K, V>>

Folds a flow of EntryEvents into a flow of Map.

@JvmName(name = "runningFoldToMapSimpleEntryEvent")

fun <K : Any, V : Any> Flow<SimpleMapEvent.EntryEvent<K, V>>.runningFoldToMap(): Flow<Map<K, V>>

Folds a flow of EntryEvents into a flow of Map.

@JvmName(name = "runningFoldToMapMapEvent")

fun <K : Any, V : Any> Flow<MapEvent<K, V>>.runningFoldToMap(emitPartials: Boolean = false): Flow<Map<K, V>>

@JvmName(name = "runningFoldToMapSimpleMapEvent")

fun <K : Any, V : Any> Flow<SimpleMapEvent<K, V>>.runningFoldToMap(emitPartials: Boolean = false): Flow<Map<K, V>>

Folds a flow of MapEvents into a flow of Map.

runningFoldToSet

fun <V : Any> Flow<SetEvent<V>>.runningFoldToSet(emitPartials: Boolean = false, relaxed: Boolean = false): Flow<PersistentSet<V>>

This method takes a flow of SetEvent and transforms it into a flow of Set. It uses a running fold operation to maintain the state of the set and emit the updated set whenever a relevant event is received.

shareInCompleting

fun <T> Flow<T>.shareInCompleting(scope: CoroutineScope, started: SharingStarted, replay: Int = 0): CompletingSharedFlow<T>

Similar to shareIn, but completions and errors are also propagated to the downstream subscribers.

throttleLatest

fun <T : Any> Flow<T>.throttleLatest(timeMillis: Long): Flow<T>

Throttles the emission of values so that they are at least timeMillis apart. Drops the older value(s) if two or more values are emitted within the time period. The first element is emitted immediately.

timeoutFirst

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

If the upstream emits no first event within duration it will emit an TimeoutException error.

fun <T> Flow<T>.timeoutFirst(millis: Long): Flow<T>

If the upstream emits no first event within millis milliseconds it will emit an TimeoutException error.

timeoutFirstOrDefault

fun <T, R> Flow<T>.timeoutFirstOrDefault(duration: Duration, default: R): Flow<R>

If the upstream emits no first event within duration it will emit the event default followed by all later emissions from the upstream.

fun <T, R> Flow<T>.timeoutFirstOrDefault(duration: Duration, default: () -> R): Flow<R>

If the upstream emits no first event within duration it will emit the event returned by default followed by all later emissions from the upstream.

fun <T, R> Flow<T>.timeoutFirstOrDefault(millis: Long, default: R): Flow<R>

If the upstream emits no first event within millis milliseconds it will emit the event default followed by all later emissions from the upstream.

fun <T, R> Flow<T>.timeoutFirstOrDefault(millis: Long, default: () -> R): Flow<R>

If the upstream emits no first event within millis milliseconds it will emit the event returned by default followed by all later emissions from the upstream.

timeoutFirstOrNull

fun <T> Flow<T>.timeoutFirstOrNull(duration: Duration): Flow<T?>

If the upstream emits no first event within duration it will emit null followed by all later emissions from the upstream.

fun <T> Flow<T>.timeoutFirstOrNull(millis: Long): Flow<T?>

If the upstream emits no first event within millis milliseconds it will emit null followed by all later emissions from the upstream.

toEvents

fun <V : Any> Flow<Set<V>>.toEvents(): Flow<SetEvent<V>>

Converts a Flow of Set to a stream of events representing the changes between consecutive sets.

toFlowMapIn

suspend fun <K : Any, V : Any> Flow<MapEvent<K, V>>.toFlowMapIn(scope: CoroutineScope): FlowMap<K, V>

@JvmName(name = "simpleToFlowMapIn")

suspend fun <K : Any, V : Any> Flow<SimpleMapEvent<K, V>>.toFlowMapIn(scope: CoroutineScope): FlowMap<K, V>

tryEmit

fun tryEmit(value: T): Boolean