refactor: onExit is flaky, explicitly retrieving result from process …

…handler once process is done
v47-io · May 10, 2024 · 38cef52 · 38cef52
1 parent ff485f3
commit 38cef52
Show file tree

Hide file tree

Showing 7 changed files with 48 additions and 57 deletions.
diff --git a/src/main/kotlin/io/v47/jaffree/ffmpeg/FFmpegProcessHandler.kt b/src/main/kotlin/io/v47/jaffree/ffmpeg/FFmpegProcessHandler.kt
@@ -65,13 +65,13 @@ internal class FFmpegProcessHandler(
             appendOrLogLine(ffmpegLogger, line.trim())
     }
 
-    override fun onExit(exitCode: Int) {
+    override fun getResult(exitCode: Int): Result<FFmpegResult> {
         processLastLogMessage(ffmpegLogger, ::handleLogMessage)
 
-        if (exitCode != 0 && finalErrorMessage != null)
-            finishExceptionally(JaffreeException(finalErrorMessage))
+        return if (exitCode != 0 && finalErrorMessage != null)
+            Result.failure(JaffreeException(finalErrorMessage))
         else
-            finish(possibleResult ?: FFmpegResult(null, null, null, null, null, null))
+            Result.success(possibleResult ?: FFmpegResult(null, null, null, null, null, null))
     }
 
     private fun handleLogMessage(logLevel: LogLevel, message: String) {

diff --git a/src/main/kotlin/io/v47/jaffree/ffprobe/FFprobeProcessHandler.kt b/src/main/kotlin/io/v47/jaffree/ffprobe/FFprobeProcessHandler.kt
@@ -56,26 +56,24 @@ internal class FFprobeProcessHandler(
         contentWritten = true
     }
 
-    override fun onExit(exitCode: Int) {
+    override fun getResult(exitCode: Int): Result<FFprobeResult> {
         stdOutputWriter.close()
         stdOutputContent.close()
 
         processLastLogMessage(ffprobeLogger)
 
-        if (exitCode != 0 && finalErrorMessage != null) {
-            finishExceptionally(JaffreeException(finalErrorMessage))
-            return
-        }
+        if (exitCode != 0 && finalErrorMessage != null)
+            return Result.failure(JaffreeException(finalErrorMessage))
 
-        ByteArrayInputStream(stdOutputContent.toByteArray()).use { input ->
-            runCatching {
-                ffprobeLogger.debug("Reading probe data using {} parser", parser.formatName)
+        ffprobeLogger.debug("Reading probe data using {} parser", parser.formatName)
 
+        return ByteArrayInputStream(stdOutputContent.toByteArray()).use { input ->
+            runCatching {
                 parser.parse(input)
-            }.onFailure { x ->
-                finishExceptionally(JaffreeException("Failed to parse probe data", x))
-            }.onSuccess { probeData ->
-                finish(FFprobeResult(probeData))
+            }.recoverCatching { x ->
+                throw JaffreeException("Failed to parse probe data", x)
+            }.map { probeData ->
+                FFprobeResult(probeData)
             }
         }
     }

diff --git a/src/main/kotlin/io/v47/jaffree/process/DelegatingProcessHandler.kt b/src/main/kotlin/io/v47/jaffree/process/DelegatingProcessHandler.kt
@@ -22,10 +22,6 @@ import java.nio.ByteBuffer
 internal class DelegatingProcessHandler(
     private val delegate: JaffreeProcessHandler<*>,
 ) : DefaultProcessHandler() {
-    override fun onExit(exitCode: Int) {
-        delegate.onExit(exitCode)
-    }
-
     override fun onStdout(buffer: ByteBuffer, closed: Boolean) {
         delegate.onStdout(buffer, closed)
     }

diff --git a/src/main/kotlin/io/v47/jaffree/process/JaffreeProcessHandler.kt b/src/main/kotlin/io/v47/jaffree/process/JaffreeProcessHandler.kt
@@ -25,7 +25,6 @@ internal interface JaffreeProcessHandler<R> {
 
     fun onStderr(buffer: ByteBuffer, closed: Boolean)
     fun onStdout(buffer: ByteBuffer, closed: Boolean)
-    fun onExit(exitCode: Int)
 
-    fun await(): Result<R>
+    fun getResult(exitCode: Int): Result<R>
 }
diff --git a/src/main/kotlin/io/v47/jaffree/process/LinesProcessHandler.kt b/src/main/kotlin/io/v47/jaffree/process/LinesProcessHandler.kt
@@ -109,19 +109,6 @@ internal abstract class LinesProcessHandler<R> : JaffreeProcessHandler<R> {
             additionalAction?.invoke(logLevel, message)
         }
     }
-
-    override fun await(): Result<R> =
-        runCatching {
-            result.get()
-        }
-
-    protected fun finish(value: R) {
-        result.complete(value)
-    }
-
-    protected fun finishExceptionally(ex: Exception) {
-        result.completeExceptionally(ex)
-    }
 }
 
 private fun addToBytes(source: ByteBuffer, target: MutableList<ByteArray>): Boolean {

diff --git a/src/main/kotlin/io/v47/jaffree/process/ProcessRunner.kt b/src/main/kotlin/io/v47/jaffree/process/ProcessRunner.kt
@@ -80,31 +80,41 @@ internal class ProcessRunner<T>(
                 val nuProcessBuilder = NuProcessBuilder(actualProcessHandler, command)
                 nuProcessBuilder.environment()["AV_LOG_FORCE_NOCOLOR"] = "1"
 
-                val process = nuProcessBuilder.start()
-                    ?: throw JaffreeAbnormalExitException("Process failed to start", emptyList())
-
-                processAccess.process = process
-
-                logger.debug("[{}] Waiting for process to finish", execTag)
-                val status = process.waitFor(0, TimeUnit.SECONDS)
-
-                logger.info("[{}] Process finished with status: {}", execTag, status)
-
-                processAccess.process = null
+                val exitCode: Int
+                val result: Result<T>
+
+                try {
+                    val process = nuProcessBuilder.start()
+                        ?: throw JaffreeAbnormalExitException(
+                            "Process failed to start",
+                            emptyList()
+                        )
+
+                    processAccess.process = process
+
+                    logger.debug("[{}] Waiting for process to finish", execTag)
+                    exitCode = process.waitFor(0, TimeUnit.SECONDS)
+                    result = processHandler.getResult(exitCode)
+                } finally {
+                    processAccess.process = null
+
+                    helpers.forEach {
+                        (it as? ProcessHelper)?.close()
+                    }
 
-                helpers.forEach {
-                    (it as? ProcessHelper)?.close()
-                }
+                    helperFutures.forEach {
+                        if (!it.isDone)
+                            it.cancel(true)
 
-                helperFutures.forEach {
-                    it.get()
+                        it.get()
+                    }
                 }
 
-                val result = processHandler.await()
+                logger.info("[{}] Process finished with status: {}", execTag, exitCode)
 
-                if (status != 0)
+                if (exitCode != 0)
                     throw JaffreeAbnormalExitException(
-                        errorExceptionMessage(status),
+                        errorExceptionMessage(exitCode),
                         processHandler.errorLogMessages
                     ).also {
                         if (result.isFailure)

diff --git a/src/main/kotlin/io/v47/jaffree/version/VersionInfoProcessHandler.kt b/src/main/kotlin/io/v47/jaffree/version/VersionInfoProcessHandler.kt
@@ -43,9 +43,10 @@ internal class VersionInfoProcessHandler : LinesProcessHandler<VersionInfo>() {
         }
     }
 
-    override fun onExit(exitCode: Int) {
-        finish(createVersionInfo())
-    }
+    override fun getResult(exitCode: Int) =
+        runCatching {
+            createVersionInfo()
+        }
 
     private fun createVersionInfo(): VersionInfo {
         val vm = versionMatch