feat(ios): proper implementation of output idle timeout

Author: Malinskiy

core/src/main/kotlin/com/malinskiy/marathon/Marathon.kt

@@ -124,7 +124,7 @@ class Marathon(
                     throw ConfigurationException("Unknown test parser type for ${testParser::class}, should inherit from either ${LocalTestParser::class.simpleName} or ${RemoteTestParser::class.simpleName}")
                 }
             }
-        }.let { println("Took $it ms") }
+        }.let { log.debug { "Parsing took $it ms" } }
 
         usageTracker.meta(
             version = BuildConfig.VERSION, releaseMode = BuildConfig.RELEASE_MODE, vendor = when (configuration.vendorConfiguration) {

vendor/vendor-ios/src/main/kotlin/com/malinskiy/marathon/ios/cmd/BaseCommand.kt

@@ -1,11 +1,17 @@
 package com.malinskiy.marathon.ios.cmd
 
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableJob
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.onClosed
+import kotlinx.coroutines.channels.onFailure
+import kotlinx.coroutines.channels.onSuccess
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.supervisorScope
 import kotlinx.coroutines.withContext
@@ -19,29 +25,37 @@ abstract class BaseCommand(
     override suspend fun await(): CommandResult = withContext(Dispatchers.IO) {
         val deferredStdout = supervisorScope {
             async(job) {
-                val stdoutBuffer = mutableListOf<String>()
-                for (line in stdout) {
-                    stdoutBuffer.add(line)
+                val buffer = mutableListOf<String>()
+                while(true) {
+                    val channelResult = stdout.receiveCatching()
+                    channelResult.onSuccess { buffer.add(it) }
+                    channelResult.onClosed { if (it != null) cancel(CancellationException("Channel closed", it)) }
+                    channelResult.onFailure { if (it != null) cancel(CancellationException("Channel failed", it)) }
+
+                    if (!channelResult.isSuccess) break
                 }
-                stdoutBuffer
+                buffer
             }
         }
 
         val deferredStderr = supervisorScope {
             async(job) {
-                val stderrBuffer = mutableListOf<String>()
-                for (line in stderr) {
-                    stderrBuffer.add(line)
+                val buffer = mutableListOf<String>()
+                while(true) {
+                    val channelResult = stderr.receiveCatching()
+                    channelResult.onSuccess { buffer.add(it) }
+                    channelResult.onClosed { if (it != null) cancel(CancellationException("Channel closed", it)) }
+                    channelResult.onFailure { if (it != null) cancel(CancellationException("Channel failed", it)) }
+
+                    if (!channelResult.isSuccess) break
                 }
-                stderrBuffer
+                buffer
             }
         }
 
-        val out = deferredStdout.await()
-        val err = deferredStderr.await()
-        val exitCode = exitCode.await()
+        val (out, err, exitCode) = awaitAll(deferredStdout, deferredStderr, exitCode)
 
-        CommandResult(out, err, exitCode)
+        CommandResult(out as List<String>, err as List<String>, exitCode as Int)
     }
 
     override suspend fun drain() {

vendor/vendor-ios/src/main/kotlin/com/malinskiy/marathon/ios/cmd/local/KotlinProcessCommandExecutor.kt

@@ -66,7 +66,7 @@ class KotlinProcessCommandExecutor(
         }
 
         val stdout = produceLinesManually(job, process.inputStream, idleTimeout, charset, channelCapacity) { process.isAlive && !exitCode.isCompleted }
-        val stderr = produceLinesManually(job, process.errorStream, idleTimeout, charset, channelCapacity) { process.isAlive && !exitCode.isCompleted }
+        val stderr = produceLinesManually(job, process.errorStream, maxOf(idleTimeout, timeout), charset, channelCapacity) { process.isAlive && !exitCode.isCompleted }
 
 
         return KotlinProcessCommand(

vendor/vendor-ios/src/main/kotlin/com/malinskiy/marathon/ios/cmd/remote/ssh/sshj/SshjCommandExecutor.kt

@@ -55,9 +55,11 @@ class SshjCommandExecutor(
             }
         }
         val cmd = session.exec(escapedCmd)
-        
+
         val stdout = produceLinesManually(job, cmd.inputStream, idleTimeout, charset, channelCapacity) { cmd.isOpen && !cmd.isEOF }
-        val stderr = produceLinesManually(job, cmd.errorStream, idleTimeout, charset, channelCapacity) { cmd.isOpen && !cmd.isEOF }
+        //Idling on stderr should not be considered bad
+        //It would be nice to combine stdout and stderr but simple version here is enough for now
+        val stderr = produceLinesManually(job, cmd.errorStream, maxOf(idleTimeout, timeout), charset, channelCapacity) { cmd.isOpen && !cmd.isEOF }
         val exitCode: Deferred<Int?> = async(job) {
             val result = withTimeoutOrNull(timeout) {
                 cmd.suspendFor()
@@ -91,7 +93,7 @@ private fun String.isShellscaped(): Boolean {
 
 private val SHELL_SPECIAL_CHARS = setOf('"',' ','$','\'','\\','#','=','[',']','!','>','<','|',';','{','}','(',')','*','?','~','&')
 private fun String.containsShellSpecialChars(): Boolean {
-    return any { 
+    return any {
         SHELL_SPECIAL_CHARS.contains(it)
     }
 }

vendor/vendor-ios/src/main/kotlin/com/malinskiy/marathon/ios/cmd/remote/ssh/sshj/SshjCommandSession.kt

@@ -4,8 +4,6 @@ import com.malinskiy.marathon.ios.cmd.BaseCommand
 import kotlinx.coroutines.CompletableJob
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.channels.ReceiveChannel
-import kotlinx.coroutines.channels.SendChannel
-import kotlinx.coroutines.channels.trySendBlocking
 import net.schmizz.sshj.connection.channel.direct.Session
 import net.schmizz.sshj.connection.channel.direct.Signal
 import java.util.concurrent.atomic.AtomicBoolean
@@ -32,6 +30,10 @@ class SshjCommandSession(
 
     override fun close() {
         if (!closed.getAndSet(true)) {
+            if (command.isOpen) {
+                terminate()
+            }
+            command.close()
             command.join()
             super.close()
         }

vendor/vendor-ios/src/main/kotlin/com/malinskiy/marathon/ios/extensions/Reader.kt

@@ -66,6 +66,13 @@ fun CoroutineScope.produceLinesManually(
                         available > 0 -> inputStream.read(byteArray, 0, min(available, byteArray.size))
                         else -> 0
                     }
+
+                    val timeSinceLastOutputMillis = System.currentTimeMillis() - lastOutputTimeMillis
+                    if (timeSinceLastOutputMillis > idleTimeout.toMillis()) {
+                        close(TimeoutException("idle timeout $idleTimeout reached"))
+                        break
+                    }
+
                     // if there was nothing to read
                     if (count == 0) {
                         // if session received EOF or has been closed, reading stops
@@ -77,11 +84,6 @@ fun CoroutineScope.produceLinesManually(
                     } else if (count == -1) {
                         break
                     } else {
-                        val timeSinceLastOutputMillis = System.currentTimeMillis() - lastOutputTimeMillis
-                        if (timeSinceLastOutputMillis > idleTimeout.toMillis()) {
-                            close(TimeoutException("idle timeout $idleTimeout reached"))
-                            break
-                        }
                         lineBuffer.append(byteArray, count)
                         lastOutputTimeMillis = System.currentTimeMillis()
                     }