Introduce Flow.any, Flow.all, Flow.none

See Kotlin#4212

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

@@ -164,3 +164,46 @@ public fun <T> Flow<T>.chunked(size: Int): Flow<List<T>> {
         result?.let { emit(it) }
     }
 }
+
+/**
+ * Returns `true` if at least one element matches the given [predicate].
+ *
+ * This operation is *terminal*.
+ *
+ * @see Iterable.any
+ * @see Sequence.any
+ */
+@ExperimentalCoroutinesApi
+public suspend fun <T> Flow<T>.any(predicate: suspend (T) -> Boolean): Boolean = this
+    .filter { predicate(it) }
+    .map { true }
+    .onEmpty { emit(false) }
+    .first()
+
+/**
+ * Returns `true` if all elements match the given [predicate].
+ *
+ * This operation is *terminal*.
+ *
+ * Note that if the flow terminates without emitting any elements, the function returns `true` because there
+ * are no elements in it that *do not* match the predicate.
+ * See a more detailed explanation of this logic concept in ["Vacuous truth"](https://en.wikipedia.org/wiki/Vacuous_truth) article.
+ *
+ * @see Iterable.all
+ * @see Sequence.all
+ */
+@ExperimentalCoroutinesApi
+public suspend fun <T> Flow<T>.all(predicate: suspend (T) -> Boolean): Boolean =
+    count { !predicate(it) } == 0
+
+/**
+ * Returns `true` if none of the elements match the given [predicate].
+ *
+ * This operation is *terminal*.
+ *
+ * @see Iterable.none
+ * @see Sequence.none
+ */
+@ExperimentalCoroutinesApi
+public suspend fun <T> Flow<T>.none(predicate: suspend (T) -> Boolean): Boolean =
+    count { predicate(it) } == 0

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

@@ -0,0 +1,58 @@
+package kotlinx.coroutines.flow
+
+import kotlinx.coroutines.testing.*
+import kotlin.test.*
+
+class BooleanTerminationTest : TestBase() {
+    @Test
+    fun testAnyNominal() = runTest {
+        val flow = flow {
+            emit(1)
+            emit(2)
+        }
+
+        assertTrue(flow.any { it > 0 })
+        assertTrue(flow.any { it % 2 == 0 })
+        assertFalse(flow.any { it > 5 })
+    }
+
+    @Test
+    fun testAnyEmpty() = runTest {
+        assertFalse(emptyFlow<Int>().any { it > 0 })
+    }
+
+    @Test
+    fun testAllNominal() = runTest {
+        val flow = flow {
+            emit(1)
+            emit(2)
+        }
+
+        assertTrue(flow.all { it > 0 })
+        assertFalse(flow.all { it % 2 == 0 })
+        assertFalse(flow.all { it > 5 })
+    }
+
+    @Test
+    fun testAllEmpty() = runTest {
+        assertTrue(emptyFlow<Int>().all { it > 0 })
+    }
+
+    @Test
+    fun testNoneNominal() = runTest {
+        val flow = flow {
+            emit(1)
+            emit(2)
+        }
+
+        assertFalse(flow.none { it > 0 })
+        assertFalse(flow.none { it % 2 == 0 })
+        assertTrue(flow.none { it > 5 })
+    }
+
+    @Test
+    fun testNoneEmpty() = runTest {
+        assertTrue(emptyFlow<Int>().none { it > 0 })
+    }
+
+}