From 103e3d5e2b64c6e6f3236b266e3948bbfc2d90ce Mon Sep 17 00:00:00 2001 From: Vsevolod Tolstopyatov Date: Wed, 20 Oct 2021 12:04:41 +0300 Subject: [PATCH] Introduce SharedFlow collect overload and override that return Nothing (#2803) * Introduce SharedFlow collect overload and override that return Nothing * Override will ensure the proper implementation of the interface * collect extension is added as the very basic lint helper Fixes #2789 Fixes #2502 Co-authored-by: dkhalanskyjb <52952525+dkhalanskyjb@users.noreply.github.com> --- .../api/kotlinx-coroutines-core.api | 2 ++ .../common/src/flow/SharedFlow.kt | 15 ++++++++++++++- .../common/src/flow/StateFlow.kt | 2 +- .../common/src/flow/terminal/Collect.kt | 13 +++++++++++++ 4 files changed, 30 insertions(+), 2 deletions(-) diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api index 495d11a804..546bfe3507 100644 --- a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api +++ b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api @@ -902,6 +902,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; @@ -1063,6 +1064,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 3424b38027..41d05a6868 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 used directly. To emit values from a shared flow into a specific collector, either `collector.emitAll(flow)` or `collect { ... }` extension + * should be used. + * + * **A shared flow never completes**. A call to [Flow.collect] or any other terminal operator + * on a shared flow never completes normally. + * + * @see [Flow.collect] + */ + @InternalCoroutinesApi + override suspend fun collect(collector: FlowCollector): Nothing } /** @@ -344,7 +357,7 @@ internal open class SharedFlowImpl( get() = buffer!!.getBufferAt(replayIndex + replaySize - 1) as T @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 53770dc91e..be6cbd6bbd 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 771f8332c3..91d410b9a2 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 the 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, only different in the return type + * so that any code below `collect` produces a 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.