Add parametrized Flow#filterIsInstance extension (#3464)

Fixes #3240

Author: IgnatBeresnev

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

@@ -973,6 +973,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun emitAll (Lkotlinx/coroutines/flow/FlowCollector;Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun emptyFlow ()Lkotlinx/coroutines/flow/Flow;
 	public static final fun filter (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun filterIsInstance (Lkotlinx/coroutines/flow/Flow;Lkotlin/reflect/KClass;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun filterNot (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun filterNotNull (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun first (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;

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

@@ -11,6 +11,7 @@ package kotlinx.coroutines.flow
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.internal.*
 import kotlin.jvm.*
+import kotlin.reflect.*
 import kotlinx.coroutines.flow.internal.unsafeFlow as flow
 import kotlinx.coroutines.flow.unsafeTransform as transform
 
@@ -34,6 +35,11 @@ public inline fun <T> Flow<T>.filterNot(crossinline predicate: suspend (T) -> Bo
 @Suppress("UNCHECKED_CAST")
 public inline fun <reified R> Flow<*>.filterIsInstance(): Flow<R> = filter { it is R } as Flow<R>
 
+/**
+ * Returns a flow containing only values that are instances of the given [klass].
+ */
+public fun <R : Any> Flow<*>.filterIsInstance(klass: KClass<R>): Flow<R> = filter { klass.isInstance(it) } as Flow<R>
+
 /**
  * Returns a flow containing only values of the original flow that are not null.
  */

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

@@ -28,6 +28,33 @@ class FilterTrivialTest : TestBase() {
         assertEquals("value", flow.filterIsInstance<String>().single())
     }
 
+    @Test
+    fun testParametrizedFilterIsInstance() = runTest {
+        val flow = flowOf("value", 2.0)
+        assertEquals(2.0, flow.filterIsInstance(Double::class).single())
+        assertEquals("value", flow.filterIsInstance(String::class).single())
+    }
+
+    @Test
+    fun testSubtypesFilterIsInstance() = runTest {
+        open class Super
+        class Sub : Super()
+
+        val flow = flowOf(Super(), Super(), Super(), Sub(), Sub(), Sub())
+        assertEquals(6, flow.filterIsInstance<Super>().count())
+        assertEquals(3, flow.filterIsInstance<Sub>().count())
+    }
+
+    @Test
+    fun testSubtypesParametrizedFilterIsInstance() = runTest {
+        open class Super
+        class Sub : Super()
+
+        val flow = flowOf(Super(), Super(), Super(), Sub(), Sub(), Sub())
+        assertEquals(6, flow.filterIsInstance(Super::class).count())
+        assertEquals(3, flow.filterIsInstance(Sub::class).count())
+    }
+
     @Test
     fun testFilterIsInstanceNullable() = runTest {
         val flow = flowOf(1, 2, null)
@@ -40,4 +67,10 @@ class FilterTrivialTest : TestBase() {
         val sum = emptyFlow<Int>().filterIsInstance<Int>().sum()
         assertEquals(0, sum)
     }
+
+    @Test
+    fun testEmptyFlowParametrizedIsInstance() = runTest {
+        val sum = emptyFlow<Int>().filterIsInstance(Int::class).sum()
+        assertEquals(0, sum)
+    }
 }