@FlowPreview class MulticastSubject<T> : 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.
bufferSize - the number of items to buffer until consumers arrive
Parameters
MulticastSubject(expectedCollectors: Int)
A subject implementation that awaits a certain number of collectors to start consuming, then allows the producer side to deliver items to them. |
val collectors: AtomicReference<Array<ResumableCollector<T>>> |
|
val producer: Resumable |
|
val remainingCollectors: AtomicInteger |
|
var terminated: Throwable? |
suspend fun collectSafely(collector: FlowCollector<T>): Unit |
|
fun collectorCount(): Int
Returns the number of collectors waiting for data. |
|
suspend fun complete(): Unit
Indicate no further items will be produced. |
|
suspend fun emit(value: T): Unit |
|
suspend fun emitError(ex: Throwable): Unit
Signal an Throwable to the collector. |
|
fun hasCollectors(): Boolean
Returns true if this subject has collectors waiting for data. |
val DONE: Throwable |
|
val EMPTY: Array<ResumableCollector<Any>> |
|
val TERMINATED: Array<ResumableCollector<Any>> |
fun <T> Flow<T>.concatWith(other: Flow<T>): Flow<T>
Emit values from the upstream followed by values from the other flow. |
|
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. |
|
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. 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. |
|
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. |
|
fun <T> Flow<T>.onBackpressurureDrop(): Flow<T>
Drops items from the upstream when the downstream is not ready to receive them. |
|
fun <T> Flow<T>.parallel(parallelism: Int, runOn: (Int) -> CoroutineDispatcher): ParallelFlow<T>
Consumes the upstream and dispatches individual items to a parrallel rail of the parallel flow for further consumption. |
|
fun <T, R> Flow<T>.publish(transform: suspend (Flow<T>) -> Flow<R>): Flow<R>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. |
|
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. 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. 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. fun <T, R> Flow<T>.replay(maxSize: Int, maxTime: Long, unit: TimeUnit, transform: suspend (Flow<T>) -> Flow<R>): Flow<R>fun <T, R> Flow<T>.replay(maxSize: Int, maxTime: Long, unit: TimeUnit, timeSource: (TimeUnit) -> Long, 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. |
|
fun <T> Flow<T>.startCollectOn(dispatcher: CoroutineDispatcher): Flow<T>
Stats collecting the upstream on the specified dispatcher. |
|
fun <T, U> Flow<T>.takeUntil(other: Flow<U>): Flow<T>
Consumes the main source until the other source emits an item or completes. |
|
fun <T> Flow<T>.toList(): Flow<List<T>>
Collects all items of the upstream into a list. |