diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api index 50bfb60d62..5f7607c581 100644 --- a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api +++ b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api @@ -900,6 +900,7 @@ public final class kotlinx/coroutines/flow/FlowKt { public static final fun channelFlow (Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow; public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; + public static final fun collect (Lkotlinx/coroutines/flow/SharedFlow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; public static final fun collectIndexed (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; public static final fun collectLatest (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; public static final fun combine (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow; @@ -1061,6 +1062,7 @@ public abstract interface class kotlinx/coroutines/flow/MutableStateFlow : kotli } public abstract interface class kotlinx/coroutines/flow/SharedFlow : kotlinx/coroutines/flow/Flow { + public abstract fun collect (Lkotlinx/coroutines/flow/FlowCollector;Lkotlin/coroutines/Continuation;)Ljava/lang/Object; public abstract fun getReplayCache ()Ljava/util/List; } diff --git a/kotlinx-coroutines-core/common/src/flow/SharedFlow.kt b/kotlinx-coroutines-core/common/src/flow/SharedFlow.kt index 9bcf088e95..bf1c90bd48 100644 --- a/kotlinx-coroutines-core/common/src/flow/SharedFlow.kt +++ b/kotlinx-coroutines-core/common/src/flow/SharedFlow.kt @@ -129,6 +129,19 @@ public interface SharedFlow : Flow { * A snapshot of the replay cache. */ public val replayCache: List + + /** + * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. + * This method should never be implemented or used directly. To collect shared flow into a specific collector, either `collector.emitAll(flow)` or `collect { ... }` extension + * should be used. + * + * **Shared flow never completes**. A call to [Flow.collect] or any other terminal operator + * on a shared flow never complete normally. + * + * @see [Flow.collect] + */ + @InternalCoroutinesApi + override suspend fun collect(collector: FlowCollector): Nothing } /** @@ -335,7 +348,7 @@ private class SharedFlowImpl( } @Suppress("UNCHECKED_CAST") - override suspend fun collect(collector: FlowCollector) { + override suspend fun collect(collector: FlowCollector): Nothing { val slot = allocateSlot() try { if (collector is SubscribedFlowCollector) collector.onSubscription() diff --git a/kotlinx-coroutines-core/common/src/flow/StateFlow.kt b/kotlinx-coroutines-core/common/src/flow/StateFlow.kt index 9e82e78771..bb79363230 100644 --- a/kotlinx-coroutines-core/common/src/flow/StateFlow.kt +++ b/kotlinx-coroutines-core/common/src/flow/StateFlow.kt @@ -380,7 +380,7 @@ private class StateFlowImpl( throw UnsupportedOperationException("MutableStateFlow.resetReplayCache is not supported") } - override suspend fun collect(collector: FlowCollector) { + override suspend fun collect(collector: FlowCollector): Nothing { val slot = allocateSlot() try { if (collector is SubscribedFlowCollector) collector.onSubscription() diff --git a/kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt b/kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt index d26839f9ea..91a94eacf5 100644 --- a/kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt +++ b/kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt @@ -73,6 +73,19 @@ public suspend inline fun Flow.collect(crossinline action: suspend (value override suspend fun emit(value: T) = action(value) }) +/** + * Terminal flow operator that collects the given [SharedFlow] with a provided [action]. + * If any exception occurs during collect or in the provided flow, this exception is rethrown from this method. + * + * This is a counterpart of a regular [Flow.collect] extension with the only difference in return type, + * so any code below `collect` will produce compilation warning. + */ +public suspend inline fun SharedFlow.collect(crossinline action: suspend (value: T) -> Unit): Nothing { + collect(object : FlowCollector { + override suspend fun emit(value: T) = action(value) + }) +} + /** * Terminal flow operator that collects the given flow with a provided [action] that takes the index of an element (zero-based) and the element. * If any exception occurs during collect or in the provided flow, this exception is rethrown from this method.