Skip to content

Commit 53669a8

Browse files
committed
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
1 parent 074cc3f commit 53669a8

File tree

4 files changed

+30
-2
lines changed

4 files changed

+30
-2
lines changed

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

+2
Original file line numberDiff line numberDiff line change
@@ -900,6 +900,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
900900
public static final fun channelFlow (Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
901901
public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
902902
public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
903+
public static final fun collect (Lkotlinx/coroutines/flow/SharedFlow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
903904
public static final fun collectIndexed (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
904905
public static final fun collectLatest (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
905906
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
10611062
}
10621063

10631064
public abstract interface class kotlinx/coroutines/flow/SharedFlow : kotlinx/coroutines/flow/Flow {
1065+
public abstract fun collect (Lkotlinx/coroutines/flow/FlowCollector;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
10641066
public abstract fun getReplayCache ()Ljava/util/List;
10651067
}
10661068

kotlinx-coroutines-core/common/src/flow/SharedFlow.kt

+14-1
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,19 @@ public interface SharedFlow<out T> : Flow<T> {
129129
* A snapshot of the replay cache.
130130
*/
131131
public val replayCache: List<T>
132+
133+
/**
134+
* Accepts the given [collector] and [emits][FlowCollector.emit] values into it.
135+
* This method should never be implemented or used directly. To collect shared flow into a specific collector, either `collector.emitAll(flow)` or `collect { ... }` extension
136+
* should be used.
137+
*
138+
* **Shared flow never completes**. A call to [Flow.collect] or any other terminal operator
139+
* on a shared flow never complete normally.
140+
*
141+
* @see [Flow.collect]
142+
*/
143+
@InternalCoroutinesApi
144+
override suspend fun collect(collector: FlowCollector<T>): Nothing
132145
}
133146

134147
/**
@@ -335,7 +348,7 @@ private class SharedFlowImpl<T>(
335348
}
336349

337350
@Suppress("UNCHECKED_CAST")
338-
override suspend fun collect(collector: FlowCollector<T>) {
351+
override suspend fun collect(collector: FlowCollector<T>): Nothing {
339352
val slot = allocateSlot()
340353
try {
341354
if (collector is SubscribedFlowCollector) collector.onSubscription()

kotlinx-coroutines-core/common/src/flow/StateFlow.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -380,7 +380,7 @@ private class StateFlowImpl<T>(
380380
throw UnsupportedOperationException("MutableStateFlow.resetReplayCache is not supported")
381381
}
382382

383-
override suspend fun collect(collector: FlowCollector<T>) {
383+
override suspend fun collect(collector: FlowCollector<T>): Nothing {
384384
val slot = allocateSlot()
385385
try {
386386
if (collector is SubscribedFlowCollector) collector.onSubscription()

kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt

+13
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,19 @@ public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value
7373
override suspend fun emit(value: T) = action(value)
7474
})
7575

76+
/**
77+
* Terminal flow operator that collects the given [SharedFlow] with a provided [action].
78+
* If any exception occurs during collect or in the provided flow, this exception is rethrown from this method.
79+
*
80+
* This is a counterpart of a regular [Flow.collect] extension with the only difference in return type,
81+
* so any code below `collect` will produce compilation warning.
82+
*/
83+
public suspend inline fun <T> SharedFlow<T>.collect(crossinline action: suspend (value: T) -> Unit): Nothing {
84+
collect(object : FlowCollector<T> {
85+
override suspend fun emit(value: T) = action(value)
86+
})
87+
}
88+
7689
/**
7790
* Terminal flow operator that collects the given flow with a provided [action] that takes the index of an element (zero-based) and the element.
7891
* If any exception occurs during collect or in the provided flow, this exception is rethrown from this method.

0 commit comments

Comments
 (0)