Introduce Stream.consumeAsFlow

    * Even though java.lang.Stream is collected rather than consumed, collectAsFlow will clash with Flow terminology where collect is a terminal operator
    * Close the stream in the end of collection despite the fact that regular terminal operations don't do that. We are already in suspending world (empty close() call won't make any difference in a common case) and "consume" implies closing the underlying resource (note that we already do it for channels)
    * Remove obsolete examples from the module

Fixes #1601

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-jdk8.txt

@@ -7,6 +7,10 @@ public final class kotlinx/coroutines/future/FutureKt {
 	public static synthetic fun future$default (Lkotlinx/coroutines/CoroutineScope;Lkotlin/coroutines/CoroutineContext;Lkotlinx/coroutines/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
 }
 
+public final class kotlinx/coroutines/stream/StreamKt {
+	public static final fun consumeAsFlow (Ljava/util/stream/Stream;)Lkotlinx/coroutines/flow/Flow;
+}
+
 public final class kotlinx/coroutines/time/TimeKt {
 	public static final fun delay (Ljava/time/Duration;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun onTimeout (Lkotlinx/coroutines/selects/SelectBuilder;Ljava/time/Duration;Lkotlin/jvm/functions/Function1;)V

integration/kotlinx-coroutines-jdk8/src/stream/Stream.kt

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.stream
+
+import kotlinx.atomicfu.*
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import java.util.stream.*
+
+/**
+ * Represents the given stream as a flow and [closes][Stream.close] the stream afterwards.
+ * The resulting flow can be [collected][Flow.collect] only once
+ * and throws [IllegalStateException] when trying to collect it more than once.
+ */
+public fun <T> Stream<T>.consumeAsFlow(): Flow<T> = StreamFlow(this)
+
+private class StreamFlow<T>(private val stream: Stream<T>) : Flow<T> {
+    private val consumed = atomic(false)
+
+    @InternalCoroutinesApi
+    override suspend fun collect(collector: FlowCollector<T>) {
+        if (!consumed.compareAndSet(false, true)) error("Stream.consumeAsFlow can be collected only once")
+        try {
+            for (value in stream.iterator()) {
+                collector.emit(value)
+            }
+        } finally {
+            stream.close()
+        }
+    }
+}

integration/kotlinx-coroutines-jdk8/test/examples/CancelFuture-example.kt

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.stream
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import org.junit.Test
+import java.lang.IllegalStateException
+import kotlin.test.*
+
+class ConsumeAsFlowTest : TestBase() {
+
+    @Test
+    fun testCollect() = runTest {
+        val list = listOf(1, 2, 3)
+        assertEquals(list, list.stream().consumeAsFlow().toList())
+    }
+
+    @Test
+    fun testCollectInvokesClose() = runTest {
+        val list = listOf(3, 4, 5)
+        expect(1)
+        assertEquals(list, list.stream().onClose { expect(2) }.consumeAsFlow().toList())
+        finish(3)
+    }
+
+    @Test
+    fun testCollectTwice() = runTest {
+        val list = listOf(2, 3, 9)
+        val flow = list.stream().onClose { expect(2) } .consumeAsFlow()
+        expect(1)
+        assertEquals(list, flow.toList())
+        assertFailsWith<IllegalStateException> { flow.collect() }
+        finish(3)
+    }
+}