Introduce SharedFlow.collectLatest (#4454)

Fixes https://youtrack.jetbrains.com/issue/KTIJ-34450

Author: dkhalanskyjb

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

@@ -1031,6 +1031,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final synthetic fun collect (Lkotlinx/coroutines/flow/Flow;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 collectLatest (Lkotlinx/coroutines/flow/SharedFlow;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;
 	public static final fun combine (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function4;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun combine (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function5;)Lkotlinx/coroutines/flow/Flow;

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

@@ -1144,6 +1144,7 @@ final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.channels/Broadcas
 final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.channels/ReceiveChannel<#A>).kotlinx.coroutines.channels/consumeEach(kotlin/Function1<#A, kotlin/Unit>) // kotlinx.coroutines.channels/consumeEach|consumeEach@kotlinx.coroutines.channels.ReceiveChannel<0:0>(kotlin.Function1<0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
 final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/Flow<#A>).kotlinx.coroutines.flow/collect(crossinline kotlin.coroutines/SuspendFunction1<#A, kotlin/Unit>) // kotlinx.coroutines.flow/collect|collect@kotlinx.coroutines.flow.Flow<0:0>(kotlin.coroutines.SuspendFunction1<0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
 final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/Flow<#A>).kotlinx.coroutines.flow/collectIndexed(crossinline kotlin.coroutines/SuspendFunction2<kotlin/Int, #A, kotlin/Unit>) // kotlinx.coroutines.flow/collectIndexed|collectIndexed@kotlinx.coroutines.flow.Flow<0:0>(kotlin.coroutines.SuspendFunction2<kotlin.Int,0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
+final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/SharedFlow<#A>).kotlinx.coroutines.flow/collectLatest(noinline kotlin.coroutines/SuspendFunction1<#A, kotlin/Unit>): kotlin/Nothing // kotlinx.coroutines.flow/collectLatest|collectLatest@kotlinx.coroutines.flow.SharedFlow<0:0>(kotlin.coroutines.SuspendFunction1<0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
 final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/SharedFlow<#A>).kotlinx.coroutines.flow/count(): kotlin/Int // kotlinx.coroutines.flow/count|count@kotlinx.coroutines.flow.SharedFlow<0:0>(){0§<kotlin.Any?>}[0]
 final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/SharedFlow<#A>).kotlinx.coroutines.flow/last(): #A // kotlinx.coroutines.flow/last|last@kotlinx.coroutines.flow.SharedFlow<0:0>(){0§<kotlin.Any?>}[0]
 final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/SharedFlow<#A>).kotlinx.coroutines.flow/toList(): kotlin.collections/List<#A> // kotlinx.coroutines.flow/toList|toList@kotlinx.coroutines.flow.SharedFlow<0:0>(){0§<kotlin.Any?>}[0]

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

@@ -111,3 +111,21 @@ public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value
     collect(object : FlowCollector<T> {
         override suspend fun emit(value: T) = action(value)
     })
+
+// -------------------- Collecting operations on a SharedFlow --------------------
+// -------------------- These mirror the operations above and are introduced when requested --------------------
+
+/**
+ * Terminal flow operator that collects the given flow with a provided [action].
+ * The crucial difference from [collect] is that when the original flow emits a new value
+ * then the [action] block for the previous value is cancelled.
+ *
+ * This is a special version of [collectLatest] for [SharedFlow].
+ * Its only difference from the usual [collectLatest] on [Flow]
+ * is that this version returns [Nothing] to indicate that it never completes.
+ * See [SharedFlow] for more details.
+ */
+public suspend inline fun <T> SharedFlow<T>.collectLatest(noinline action: suspend (value: T) -> Unit): Nothing {
+    (this as Flow<T>).collectLatest(action)
+    throw IllegalStateException("SharedFlow never completes, this call should never return.")
+}