ProcessRunner.kt

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import java.io.File
import java.io.IOException
import kotlin.coroutines.resumeWithException

@ExperimentalCoroutinesApi
suspend fun startProcess(cmd: String, workingDir: File): Process = suspendCancellableCoroutine { cont ->
    try {
        val process =
            ProcessBuilder(*cmd.split("\\s".toRegex()).toTypedArray())
                .directory(workingDir)
                .start()

        cont.invokeOnCancellation {
            println("Destroying $process due to cancellation before started.")
            process.destroy()
        }

        cont.resume(process) {
            process.destroy()
        }
    } catch (e: Throwable) {
        cont.resumeWithException(e)
    }
}

@ExperimentalCoroutinesApi
suspend fun Process.wait(): Unit = suspendCancellableCoroutine { cont ->
    cont.invokeOnCancellation {
        println("Destroying $this due to cancellation.")
        destroy()
    }

    try {
        when (val returnCode = waitFor()) {
            0 -> {
                cont.resume(Unit) { destroy() }
            }
            else -> {
                cont.resumeWithException(IOException("Process $this exits with non zero code $returnCode."))
            }
        }
    } catch (e: Throwable) {
        cont.resumeWithException(e)
    }
}

interface ProcessOutLine {
    val line: String
}

data class StdOutLine(override val line: String) : ProcessOutLine
data class StdErrLine(override val line: String) : ProcessOutLine

@ExperimentalCoroutinesApi
@FlowPreview
suspend fun runCommand(cmd: String, workingDir: File = File(".")): Flow<ProcessOutLine> = channelFlow {
    withContext(Dispatchers.IO) {
        val process = startProcess(cmd, workingDir)

        launch {
            for (line in process.inputStream.bufferedReader().lines()) {
                send(StdOutLine(line))
            }
        }

        launch {
            for (line in process.errorStream.bufferedReader().lines()) {
                send(StdErrLine(line))
            }
        }

        process.wait()
    }
}

@ExperimentalCoroutinesApi
@FlowPreview
suspend fun main(): Unit = coroutineScope {
    withTimeout(500) {
        runCommand("java -version").collect {
            println(it)
            //delay(1000)
        }
    }
}

• withTimeout(500), no delay on each line:

StdErrLine(line=openjdk version "14.0.1" 2020-04-14)
StdErrLine(line=OpenJDK Runtime Environment (build 14.0.1+7))
StdErrLine(line=OpenJDK 64-Bit Server VM (build 14.0.1+7, mixed mode, sharing))

Process finished with exit code 0

• with delay(1000) after each line consumed:

StdErrLine(line=openjdk version "14.0.1" 2020-04-14)
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 500 ms
	at kotlinx.coroutines.TimeoutKt.TimeoutCancellationException(Timeout.kt:158)
	at kotlinx.coroutines.TimeoutCoroutine.run(Timeout.kt:128)
	at kotlinx.coroutines.EventLoopImplBase$DelayedRunnableTask.run(EventLoop.common.kt:497)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:274)
	at kotlinx.coroutines.DefaultExecutor.run(DefaultExecutor.kt:68)
	at java.base/java.lang.Thread.run(Thread.java:832)

Process finished with exit code 1

• withTimeout(100):

Destroying process Process[pid=3319, exitValue="not exited"] due to cancellation.  <------------------------- finally, it's cancelled
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 100 ms
	at kotlinx.coroutines.TimeoutKt.TimeoutCancellationException(Timeout.kt:158)
	at kotlinx.coroutines.TimeoutCoroutine.run(Timeout.kt:128)
	at kotlinx.coroutines.EventLoopImplBase$DelayedRunnableTask.run(EventLoop.common.kt:497)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:274)
	at kotlinx.coroutines.DefaultExecutor.run(DefaultExecutor.kt:68)
	at java.base/java.lang.Thread.run(Thread.java:832)

Process finished with exit code 1