Fix collecting SharedFlow into a mutable collection (#3711)

Fixes #3706

Author: dkhalanskyjb

kotlinx-coroutines-core/common/src/flow/operators/Lint.kt

@@ -144,9 +144,19 @@ public inline fun <T> SharedFlow<T>.retryWhen(noinline predicate: suspend FlowCo
     level = DeprecationLevel.WARNING
 )
 @InlineOnly
-public suspend inline fun <T> SharedFlow<T>.toList(destination: MutableList<T> = ArrayList()): List<T> =
+public suspend inline fun <T> SharedFlow<T>.toList(): List<T> =
     (this as Flow<T>).toList()
 
+/**
+ * A specialized version of [Flow.toList] that returns [Nothing]
+ * to indicate that [SharedFlow] collection never completes.
+ */
+@InlineOnly
+public suspend inline fun <T> SharedFlow<T>.toList(destination: MutableList<T>): Nothing {
+    (this as Flow<T>).toList(destination)
+    throw IllegalStateException("this code is supposed to be unreachable")
+}
+
 /**
  * @suppress
  */
@@ -156,9 +166,19 @@ public suspend inline fun <T> SharedFlow<T>.toList(destination: MutableList<T> =
     level = DeprecationLevel.WARNING
 )
 @InlineOnly
-public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T> = LinkedHashSet()): Set<T> =
+public suspend inline fun <T> SharedFlow<T>.toSet(): Set<T> =
     (this as Flow<T>).toSet()
 
+/**
+ * A specialized version of [Flow.toSet] that returns [Nothing]
+ * to indicate that [SharedFlow] collection never completes.
+ */
+@InlineOnly
+public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
+    (this as Flow<T>).toSet(destination)
+    throw IllegalStateException("this code is supposed to be unreachable")
+}
+
 /**
  * @suppress
  */

kotlinx-coroutines-core/common/test/flow/operators/LintTest.kt

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2016-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow.operators
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlin.test.*
+
+class LintTest: TestBase() {
+    /**
+     * Tests that using [SharedFlow.toList] and similar functions by passing a mutable collection does add values
+     * to the provided collection.
+     */
+    @Test
+    fun testSharedFlowToCollection() = runTest {
+        val sharedFlow = MutableSharedFlow<Int>()
+        val list = mutableListOf<Int>()
+        val set = mutableSetOf<Int>()
+        val jobs = listOf(suspend { sharedFlow.toList(list) }, { sharedFlow.toSet(set) }).map {
+            launch(Dispatchers.Unconfined) { it() }
+        }
+        repeat(10) {
+            sharedFlow.emit(it)
+        }
+        jobs.forEach { it.cancelAndJoin() }
+        assertEquals((0..9).toList(), list)
+        assertEquals((0..9).toSet(), set)
+    }
+}