Reintroduce the initial proposal, better being safe than sorry

pgreze · Aug 4, 2021 · 246c46d · 246c46d
1 parent 39e153d
commit 246c46d
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 21 deletions.
diff --git a/src/main/kotlin/com/github/pgreze/process/Process.kt b/src/main/kotlin/com/github/pgreze/process/Process.kt
@@ -7,6 +7,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.map
@@ -72,8 +73,12 @@ suspend fun process(
             stderr == Redirect.CAPTURE ->
                 process.errorStream
             else -> null
-        }?.lineFlow { f -> f.map { it.also { consumer(it) } }.toList() }
-            ?: emptyList()
+        }?.lineFlow { f ->
+            f.map {
+                yield()
+                it.also { consumer(it) }
+            }.toList()
+        } ?: emptyList()
     }
 
     val input = async {

diff --git a/src/test/kotlin/com/github/pgreze/process/ProcessKtTest.kt b/src/test/kotlin/com/github/pgreze/process/ProcessKtTest.kt
@@ -24,7 +24,6 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.ValueSource
 import java.io.ByteArrayOutputStream
 import java.io.File
-import java.io.InputStream
 import java.io.PrintStream
 import java.nio.file.Path
 import java.util.concurrent.TimeUnit
@@ -166,27 +165,35 @@ class ProcessKtTest {
         stdout shouldBeEqualTo OUT.toList()
     }
 
-    @Test
-    @Timeout(value = 3, unit = TimeUnit.SECONDS)
-    fun `job cancellation should destroy the process`() = runSuspendTest {
-        var visitedCancelledBlock = false
-        val job = launch(Dispatchers.IO) {
-            try {
-                val ret = process("cat") // cat without args is an endless process.
-                throw AssertionError("Process completed despite being cancelled: $ret")
-            } catch (e: CancellationException) {
-                visitedCancelledBlock = true
+    @Nested
+    @DisplayName("print to console or not")
+    inner class Cancellation {
+        @ParameterizedTest
+        @ValueSource(booleans = [true, false])
+        @Timeout(value = 3, unit = TimeUnit.SECONDS)
+        fun `job cancellation should destroy the process`(captureStdout: Boolean) = runSuspendTest {
+            var visitedCancelledBlock = false
+            val job = launch(Dispatchers.IO) {
+                try {
+                    val ret = process(
+                        "cat", // cat without args is an endless process.
+                        stdout = if (captureStdout) CAPTURE else SILENT
+                    )
+                    throw AssertionError("Process completed despite being cancelled: $ret")
+                } catch (e: CancellationException) {
+                    visitedCancelledBlock = true
+                }
             }
-        }
 
-        // Introduce delays to be sure the job was started before being cancelled.
-        delay(500L)
-        job.cancel()
-        delay(500L)
+            // Introduce delays to be sure the job was started before being cancelled.
+            delay(500L)
+            job.cancel()
+            delay(500L)
 
-        job.isCancelled shouldBeEqualTo true
-        job.isCompleted shouldBeEqualTo true
-        visitedCancelledBlock shouldBeEqualTo true
+            job.isCancelled shouldBeEqualTo true
+            job.isCompleted shouldBeEqualTo true
+            visitedCancelledBlock shouldBeEqualTo true
+        }
     }
 
     @Nested