Moved to props + fixing abandoned issue

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/CoordinatorApplication.kt

@@ -7,6 +7,8 @@ import no.iktdev.exfl.coroutines.CoroutinesDefault
 import no.iktdev.exfl.coroutines.CoroutinesIO
 import no.iktdev.exfl.observable.Observables
 import no.iktdev.mediaprocessing.coordinator.config.ExecutablesConfig
+import no.iktdev.mediaprocessing.coordinator.config.ProcesserClientProperties
+import no.iktdev.mediaprocessing.shared.common.configs.StreamItConfig
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.EventRegistry
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.TaskRegistry
 import no.iktdev.mediaprocessing.shared.common.getAppVersion
@@ -58,7 +60,9 @@ open class ApplicationConfiguration() {
 @Configuration
 @EnableConfigurationProperties(
     value = [
-        ExecutablesConfig::class
+        ExecutablesConfig::class,
+        StreamItConfig::class,
+        ProcesserClientProperties::class
     ]
 )
 class CoordinatorConfig

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/ProcesserClient.kt

@@ -0,0 +1,23 @@
+package no.iktdev.mediaprocessing.coordinator
+
+import org.springframework.stereotype.Component
+import org.springframework.web.reactive.function.client.WebClient
+import reactor.core.publisher.Mono
+
+@Component
+class ProcesserClient(
+    private val webClient: WebClient
+) {
+
+    fun fetchLog(path: String): Mono<String> =
+        webClient.get()
+            .uri { it.path("/state/log").queryParam("path", path).build() }
+            .retrieve()
+            .bodyToMono(String::class.java)
+
+    fun ping(): Mono<String> =
+        webClient.get()
+            .uri("/actuator/health")
+            .retrieve()
+            .bodyToMono(String::class.java)
+}

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/config/ProcesserClientConfig.kt

@@ -0,0 +1,25 @@
+package no.iktdev.mediaprocessing.coordinator.config
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.web.reactive.function.client.WebClient
+
+@Configuration
+class ProcesserWebClientConfig {
+
+    @Bean
+    fun processerWebClient(
+        builder: WebClient.Builder,
+        props: ProcesserClientProperties
+    ): WebClient =
+        builder
+            .baseUrl(props.baseUrl)
+            .build()
+}
+
+
+@ConfigurationProperties(prefix = "processer")
+data class ProcesserClientProperties(
+    val baseUrl: String
+)

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/dto/translate/CoordinatorTaskTransferDto.kt

@@ -34,6 +34,6 @@ fun PersistedTask.toCoordinatorTransferDto(): CoordinatorTaskTransferDto {
         consumed = consumed,
         lastCheckIn = lastCheckIn,
         persistedAt = persistedAt,
-        abandoned = TaskLifecycleRules.isAbandoned(consumed, lastCheckIn)
+        abandoned = TaskLifecycleRules.isAbandoned(consumed, persistedAt, lastCheckIn)
     )
 }

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/services/CoordinatorHealthService.kt

@@ -31,7 +31,7 @@ class CoordinatorHealthService(
 
         // --- TASK HEALTH ---
         val abandonedTaskIds = tasks
-            .filter { TaskLifecycleRules.isAbandoned(it.consumed, it.lastCheckIn) }
+            .filter { TaskLifecycleRules.isAbandoned(it.consumed, it.persistedAt, it.lastCheckIn) }
             .map { it.taskId }
 
         val stalledTaskIds = tasks

apps/coordinator/src/main/resources/application.yml

@@ -31,7 +31,10 @@ media:
   incoming: /src/input
 
 streamit:
-  address: http://streamit.service
+  base-url: ${PROCESSER_URL:http://processer:8080}
+
+processer:
+  base-url: http://processer.service:8080
 
 executables:
   ffprobe: ffprobe

apps/coordinator/src/test/resources/application.yml

@@ -40,5 +40,8 @@ media:
 streamit:
   address: http://streamit.service
 
+processer:
+  base-url: http://processer.service:8080
+
 executables:
   ffprobe: ffprobe

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/CoordinatorClient.kt

@@ -1,6 +1,9 @@
 package no.iktdev.mediaprocessing.processer
 
+import jakarta.annotation.PostConstruct
+import mu.KotlinLogging
 import no.iktdev.mediaprocessing.ffmpeg.decoder.FfmpegDecodedProgress
+import no.iktdev.mediaprocessing.processer.config.ProcesserProperties
 import no.iktdev.mediaprocessing.shared.common.model.ProgressUpdate
 import org.springframework.http.MediaType
 import org.springframework.stereotype.Component
@@ -8,17 +11,35 @@ import org.springframework.web.reactive.function.client.WebClient
 
 @Component
 class CoordinatorClient(
+    private val processerProperties: ProcesserProperties,
     private val webClient: WebClient
 ) {
+    private val log = KotlinLogging.logger {}
+
+    @PostConstruct
+    fun pingCoordinator() {
+        if (!processerProperties.coordinatorPingOnStartup) {
+            log.info { "Coordinator ping on startup is disabled" }
+            return
+        }
+        try {
+            val result = webClient.get()
+                .uri("/actuator/health")
+                .retrieve()
+                .bodyToMono(String::class.java)
+                .block()
+
+            log.info { "Coordinator reachable. Health: $result" }
+        } catch (e: Exception) {
+            log.error(e) { "Coordinator NOT reachable at startup" }
+        }
+    }
 
     fun reportProgress(referenceId: String, taskId: String, percent: FfmpegDecodedProgress, message: String?) =
         webClient.post()
             .uri("/internal/progress")
             .contentType(MediaType.APPLICATION_JSON)
-            .bodyValue(
+            .bodyValue(ProgressUpdate(referenceId, taskId, percent, message))
-                ProgressUpdate(referenceId, taskId, percent, message)
-            )
             .retrieve()
             .toBodilessEntity()
-
 }

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/FileUtil.kt

@@ -0,0 +1,18 @@
+package no.iktdev.mediaprocessing.processer
+
+import no.iktdev.exfl.using
+import no.iktdev.mediaprocessing.processer.config.DirectoryProperties
+import no.iktdev.mediaprocessing.shared.common.configs.MediaPaths
+import org.springframework.stereotype.Component
+import java.io.File
+
+@Component
+class FileUtil(
+    private val dirs: DirectoryProperties,
+    private val mediaPaths: MediaPaths
+) {
+    fun getTemporaryStoreFile(fileName: String): File =
+        File(mediaPaths.cache).using(fileName)
+
+    fun getLogDirectory(): File = File(dirs.logs)
+}

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/LocalProgressCache.kt

@@ -0,0 +1,19 @@
+package no.iktdev.mediaprocessing.processer
+
+import no.iktdev.mediaprocessing.ffmpeg.decoder.FfmpegDecodedProgress
+import org.springframework.stereotype.Component
+import java.util.*
+import java.util.concurrent.ConcurrentHashMap
+
+@Component
+class LocalProgressCache {
+    private val cache = ConcurrentHashMap<UUID, FfmpegDecodedProgress>()
+
+    fun update(taskId: UUID, progress: FfmpegDecodedProgress) {
+        cache[taskId] = progress
+    }
+
+    fun get(taskId: UUID): FfmpegDecodedProgress? = cache[taskId]
+
+    fun getAll(): Map<UUID, FfmpegDecodedProgress> = cache.toMap()
+}

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/ProcesserApplication.kt

@@ -6,6 +6,10 @@ import no.iktdev.eventi.tasks.TaskTypeRegistry
 import no.iktdev.exfl.coroutines.CoroutinesDefault
 import no.iktdev.exfl.coroutines.CoroutinesIO
 import no.iktdev.exfl.observable.Observables
+import no.iktdev.mediaprocessing.processer.config.DirectoryProperties
+import no.iktdev.mediaprocessing.processer.config.ExecutablesConfig
+import no.iktdev.mediaprocessing.processer.config.ProcesserProperties
+import no.iktdev.mediaprocessing.shared.common.configs.MediaPaths
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.EventRegistry
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.TaskRegistry
 import no.iktdev.mediaprocessing.shared.common.getAppVersion
@@ -57,7 +61,10 @@ open class ApplicationConfiguration() {
 @Configuration
 @EnableConfigurationProperties(
     value = [
-        ExecutablesConfig::class
+        ExecutablesConfig::class,
+        DirectoryProperties::class,
+        ProcesserProperties::class,
+        MediaPaths::class
     ]
 )
 class ProcesserConfig

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/ProcesserEnv.kt

@@ -1,25 +0,0 @@
-package no.iktdev.mediaprocessing.processer
-
-import no.iktdev.exfl.using
-import java.io.File
-
-class ProcesserEnv {
-    companion object {
-        val coordinatorUrl = System.getenv("COORDINATOR_URL") ?: "http://coordinator:8080"
-
-        val ffmpeg: String = System.getenv("SUPPORTING_EXECUTABLE_FFMPEG") ?: "ffmpeg"
-        val allowOverwrite = System.getenv("ALLOW_OVERWRITE").toBoolean() ?: false
-
-        var cachedContent: File = if (!System.getenv("DIRECTORY_CONTENT_CACHE").isNullOrBlank()) File(System.getenv("DIRECTORY_CONTENT_CACHE")) else File("/src/cache")
-
-
-        val logDirectory = if (!System.getenv("LOG_DIR").isNullOrBlank()) File(System.getenv("LOG_DIR")) else
-            File("data").using("logs")
-
-        val encodeLogDirectory = logDirectory.using("encode")
-        val extractLogDirectory = logDirectory.using("extract")
-        val subtitleExtractLogDirectory = logDirectory.using("subtitles")
-
-        val fullLogging = System.getenv("FullLogging").toBoolean()
-    }
-}

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/Util.kt

@@ -1,10 +0,0 @@
-package no.iktdev.mediaprocessing.processer
-
-import no.iktdev.exfl.using
-import java.io.File
-
-object Util {
-    fun getTemporaryStoreFile(fileName: String): File {
-        return ProcesserEnv.cachedContent.using(fileName)
-    }
-}

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/config/CoordinatorClientConfig.kt

@@ -1,15 +1,17 @@
-package no.iktdev.mediaprocessing.processer
+package no.iktdev.mediaprocessing.processer.config
 
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.web.reactive.function.client.WebClient
 
 @Configuration
-class CoordinatorClientConfig {
+class CoordinatorClientConfig(
+    private val processerProperties: ProcesserProperties,
+) {
 
     @Bean
     fun coordinatorWebClient(builder: WebClient.Builder): WebClient {
-        val baseUrl = ProcesserEnv.coordinatorUrl
+        val baseUrl = processerProperties.coordinatorUrl
             ?: error("COORDINATOR_URL must be set")
 
         return builder
@@ -17,3 +19,4 @@ class CoordinatorClientConfig {
             .build()
     }
 }
+

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/config/DirectoryProperties.kt

@@ -0,0 +1,8 @@
+package no.iktdev.mediaprocessing.processer.config
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+
+@ConfigurationProperties(prefix = "directories")
+data class DirectoryProperties(
+    val logs: String,
+)

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/config/ExecutableConfig.kt

@@ -1,4 +1,4 @@
-package no.iktdev.mediaprocessing.processer
+package no.iktdev.mediaprocessing.processer.config
 
 import org.springframework.boot.context.properties.ConfigurationProperties
 

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/config/ProcesserProperties.kt

@@ -0,0 +1,10 @@
+package no.iktdev.mediaprocessing.processer.config
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+
+@ConfigurationProperties(prefix = "processer")
+data class ProcesserProperties(
+    val coordinatorUrl: String,
+    val coordinatorPingOnStartup: Boolean,
+    val allowOverwrite: Boolean
+)

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/controller/StateController.kt

@@ -0,0 +1,36 @@
+package no.iktdev.mediaprocessing.processer.controller
+
+import no.iktdev.mediaprocessing.ffmpeg.decoder.FfmpegDecodedProgress
+import no.iktdev.mediaprocessing.processer.LocalProgressCache
+import org.springframework.http.ResponseEntity
+import org.springframework.web.bind.annotation.*
+import java.io.File
+import java.util.*
+
+@RestController
+@RequestMapping("/state")
+class StateController(
+    private val localProgress: LocalProgressCache
+) {
+
+    @GetMapping("/progress")
+    fun allProgress(): Map<UUID, FfmpegDecodedProgress> =
+        localProgress.getAll()
+
+    @GetMapping("/progress/{taskId}")
+    fun progress(@PathVariable taskId: UUID): FfmpegDecodedProgress? =
+        localProgress.get(taskId)
+
+
+    @GetMapping("/log")
+    fun getLog(@RequestParam path: String): ResponseEntity<String> {
+        val file = File(path)
+        return if (file.exists()) {
+            ResponseEntity.ok(file.readText())
+        } else {
+            ResponseEntity.notFound().build()
+        }
+    }
+
+
+}

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/listeners/SubtitleTaskListener.kt

@@ -6,17 +6,22 @@ import no.iktdev.eventi.models.Task
 import no.iktdev.eventi.models.store.TaskStatus
 import no.iktdev.eventi.tasks.TaskReporter
 import no.iktdev.eventi.tasks.TaskType
+import no.iktdev.exfl.using
 import no.iktdev.mediaprocessing.ffmpeg.FFmpeg
 import no.iktdev.mediaprocessing.ffmpeg.arguments.MpegArgument
-import no.iktdev.mediaprocessing.processer.ProcesserEnv
+import no.iktdev.mediaprocessing.processer.FileUtil
-import no.iktdev.mediaprocessing.processer.Util
+import no.iktdev.mediaprocessing.processer.config.ExecutablesConfig
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.events.ProcesserExtractResultEvent
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.tasks.ExtractSubtitleTask
 import org.springframework.stereotype.Service
+import java.io.File
 import java.util.*
 
 @Service
-class SubtitleTaskListener: FfmpegTaskListener(TaskType.CPU_INTENSIVE) {
+class SubtitleTaskListener(
+    private val executableConfig: ExecutablesConfig,
+    private val fileUtil: FileUtil
+) : FfmpegTaskListener(TaskType.CPU_INTENSIVE) {
     private val log = KotlinLogging.logger {}
 
 
@@ -35,16 +40,18 @@ class SubtitleTaskListener: FfmpegTaskListener(TaskType.CPU_INTENSIVE) {
     override suspend fun onTask(task: Task): Event? {
         val taskData = task as ExtractSubtitleTask
 
-        val cachedOutFile = Util.getTemporaryStoreFile(taskData.data.outputFileName).also {
+        val cachedOutFile = fileUtil.getTemporaryStoreFile(taskData.data.outputFileName).also {
             if (!it.parentFile.exists()) {
                 it.parentFile.mkdirs()
             }
         }
 
         if (cachedOutFile.exists() && taskData.data.arguments.firstOrNull() != "-y") {
-            reporter?.publishEvent(ProcesserExtractResultEvent(
+            reporter?.publishEvent(
-                status = TaskStatus.Failed
+                ProcesserExtractResultEvent(
-            ).producedFrom(task))
+                    status = TaskStatus.Failed
+                ).producedFrom(task)
+            )
             throw IllegalStateException("${cachedOutFile.absolutePath} does already exist, and arguments does not permit overwrite")
         }
 
@@ -58,7 +65,7 @@ class SubtitleTaskListener: FfmpegTaskListener(TaskType.CPU_INTENSIVE) {
             reporter?.updateLastSeen(task.taskId)
         }
         result.run(arguments)
-        if (result.result.resultCode != 0 ) {
+        if (result.result.resultCode != 0) {
             return ProcesserExtractResultEvent(status = TaskStatus.Failed).producedFrom(task)
         }
 
@@ -85,15 +92,17 @@ class SubtitleTaskListener: FfmpegTaskListener(TaskType.CPU_INTENSIVE) {
     }
 
     override fun getFfmpeg(): FFmpeg {
-        return SubtitleFFmpeg()
+        val logDirectory = fileUtil.getLogDirectory().using("subtitles")
+        return SubtitleFFmpeg(null, executableConfig.ffmpeg, logDirectory)
     }
 
 
-    class SubtitleFFmpeg(override val listener: Listener? = null): FFmpeg(executable = ProcesserEnv.ffmpeg, logDir = ProcesserEnv.subtitleExtractLogDirectory ) {
+    class SubtitleFFmpeg(override val listener: Listener? = null, private val executablePath: String, val logDirectory: File) :
+        FFmpeg(executable = executablePath, logDir = logDirectory) {
 
         override fun onCreate() {
-            if (!ProcesserEnv.subtitleExtractLogDirectory.exists()) {
+            if (!logDir.exists()) {
-                ProcesserEnv.subtitleExtractLogDirectory.mkdirs()
+                logDir.mkdirs()
             }
         }
     }

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/listeners/VideoTaskListener.kt

@@ -6,19 +6,27 @@ import no.iktdev.eventi.models.Task
 import no.iktdev.eventi.models.store.TaskStatus
 import no.iktdev.eventi.tasks.TaskReporter
 import no.iktdev.eventi.tasks.TaskType
+import no.iktdev.exfl.using
 import no.iktdev.mediaprocessing.ffmpeg.FFmpeg
 import no.iktdev.mediaprocessing.ffmpeg.arguments.MpegArgument
 import no.iktdev.mediaprocessing.ffmpeg.decoder.FfmpegDecodedProgress
 import no.iktdev.mediaprocessing.processer.CoordinatorClient
-import no.iktdev.mediaprocessing.processer.ProcesserEnv
+import no.iktdev.mediaprocessing.processer.FileUtil
-import no.iktdev.mediaprocessing.processer.Util
+import no.iktdev.mediaprocessing.processer.LocalProgressCache
+import no.iktdev.mediaprocessing.processer.config.ExecutablesConfig
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.events.ProcesserEncodeResultEvent
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.tasks.EncodeTask
 import org.springframework.stereotype.Service
+import java.io.File
 import java.util.*
 
 @Service
-class VideoTaskListener(private var coordinatorWebClient: CoordinatorClient): FfmpegTaskListener(TaskType.CPU_INTENSIVE) {
+class VideoTaskListener(
+    private var coordinatorWebClient: CoordinatorClient,
+    private val localProgress: LocalProgressCache,
+    private val executableConfig: ExecutablesConfig,
+    private val fileUtil: FileUtil,
+) : FfmpegTaskListener(TaskType.CPU_INTENSIVE) {
     private val log = KotlinLogging.logger {}
 
     override fun getWorkerId() = "${this::class.java.simpleName}-${taskType}-${UUID.randomUUID()}"
@@ -35,15 +43,17 @@ class VideoTaskListener(private var coordinatorWebClient: CoordinatorClient): Ff
 
     override suspend fun onTask(task: Task): Event? {
         val taskData = task as EncodeTask
-        val cachedOutFile = Util.getTemporaryStoreFile(taskData.data.outputFileName).also {
+        val cachedOutFile = fileUtil.getTemporaryStoreFile(taskData.data.outputFileName).also {
             if (!it.parentFile.exists()) {
                 it.parentFile.mkdirs()
             }
         }
         if (cachedOutFile.exists() && taskData.data.arguments.firstOrNull() != "-y") {
-            reporter?.publishEvent(ProcesserEncodeResultEvent(
+            reporter?.publishEvent(
-                status = TaskStatus.Failed
+                ProcesserEncodeResultEvent(
-            ).producedFrom(task))
+                    status = TaskStatus.Failed
+                ).producedFrom(task)
+            )
             throw IllegalStateException("${cachedOutFile.absolutePath} does already exist, and arguments does not permit overwrite")
         }
 
@@ -58,12 +68,16 @@ class VideoTaskListener(private var coordinatorWebClient: CoordinatorClient): Ff
             reporter?.updateLastSeen(task.taskId)
         }
         result.run(arguments)
-        if (result.result.resultCode != 0 ) {
+        if (result.result.resultCode != 0) {
-            return ProcesserEncodeResultEvent(status = TaskStatus.Failed).producedFrom(task)
+            return ProcesserEncodeResultEvent(
+                status = TaskStatus.Failed,
+                logFile = result.logFile.absolutePath
+            ).producedFrom(task)
         }
 
         return ProcesserEncodeResultEvent(
             status = TaskStatus.Completed,
+            logFile = result.logFile.absolutePath,
             data = ProcesserEncodeResultEvent.EncodeResult(
                 cachedOutputFile = cachedOutFile.absolutePath
             )
@@ -80,12 +94,15 @@ class VideoTaskListener(private var coordinatorWebClient: CoordinatorClient): Ff
             TaskStatus.Cancelled -> "Canceled"
             else -> ""
         }
-        return ProcesserEncodeResultEvent(null, status, error = message)
+        return ProcesserEncodeResultEvent(null, null, status, error = message)
     }
 
 
     override fun getFfmpeg(): FFmpeg {
-        return VideoFFmpeg(object : FFmpeg.Listener {
+        val logDirectory = fileUtil.getLogDirectory().using("encode")
+        return VideoFFmpeg(execPath = executableConfig.ffmpeg,
+            logDirectory = logDirectory,
+            listener = object : FFmpeg.Listener {
             var lastProgress: FfmpegDecodedProgress? = null
             override fun onStarted(inputFile: String) {
             }
@@ -95,7 +112,14 @@ class VideoTaskListener(private var coordinatorWebClient: CoordinatorClient): Ff
                     coordinatorWebClient.reportProgress(
                         referenceId = it.referenceId.toString(),
                         taskId = it.taskId.toString(),
-                        percent = FfmpegDecodedProgress(100, "", lastProgress?.duration ?: "", "0", estimatedCompletion = "", estimatedCompletionSeconds = 0),
+                        percent = FfmpegDecodedProgress(
+                            100,
+                            "",
+                            lastProgress?.duration ?: "",
+                            "0",
+                            estimatedCompletion = "",
+                            estimatedCompletionSeconds = 0
+                        ),
                         ""
                     )
                 }
@@ -107,6 +131,7 @@ class VideoTaskListener(private var coordinatorWebClient: CoordinatorClient): Ff
             ) {
                 lastProgress = progress
                 currentTask?.let {
+                    localProgress.update(it.taskId, progress)
                     coordinatorWebClient.reportProgress(
                         referenceId = it.referenceId.toString(),
                         taskId = it.taskId.toString(),
@@ -120,12 +145,13 @@ class VideoTaskListener(private var coordinatorWebClient: CoordinatorClient): Ff
     }
 
 
-    class VideoFFmpeg(override val listener: Listener? = null): FFmpeg(executable = ProcesserEnv.ffmpeg, logDir = ProcesserEnv.encodeLogDirectory) {
+    class VideoFFmpeg(override val listener: Listener? = null, private val execPath: String, val logDirectory: File) :
+        FFmpeg(executable = execPath, logDir = logDirectory) {
 
         override fun onCreate() {
             super.onCreate()
-            if (!ProcesserEnv.encodeLogDirectory.exists()) {
+            if (!logDirectory.exists()) {
-                ProcesserEnv.encodeLogDirectory.mkdirs()
+                logDirectory.mkdirs()
             }
         }
     }

apps/processer/src/main/resources/application.yml

@@ -26,12 +26,17 @@ logging:
 
 
 media:
-  cache: /src/cache
+  cache: ${DIRECTORY_CONTENT_CACHE:/src/cache}
   outgoing: /src/output
   incoming: /src/input
 
-streamit:
+processer:
-  address: http://streamit.service
+  coordinator-url: ${COORDINATOR_URL:http://coordinator:8080}
+  coordinator-ping-on-startup: true
+  allow-overwrite: ${ALLOW_OVERWRITE:false}
 
 executables:
-  ffmpeg: ffmpeg
+  ffmpeg: ${SUPPORTING_EXECUTABLE_FFMPEG:ffmpeg}
+
+directories:
+  logs: ${LOG_DIR:/data/logs}

apps/processer/src/test/kotlin/no/iktdev/mediaprocessing/processer/TestUtils.kt

@@ -0,0 +1,27 @@
+package no.iktdev.mediaprocessing.processer
+
+import no.iktdev.mediaprocessing.processer.config.DirectoryProperties
+import no.iktdev.mediaprocessing.processer.config.ExecutablesConfig
+import no.iktdev.mediaprocessing.shared.common.configs.MediaPaths
+
+object TestUtils {
+    fun getFileUtil(): FileUtil {
+        val dirs = DirectoryProperties(
+            logs = "build/test-logs",
+        )
+        val mediaPaths = MediaPaths(
+            cache = "build/test-cache",
+            incoming = "build/test-input",
+            outgoing = "build/test-output"
+        )
+
+        return FileUtil(dirs, mediaPaths)
+    }
+
+    fun getExecutableConfig(): ExecutablesConfig {
+        return ExecutablesConfig(
+            ffmpeg = "ffmpeg"
+        )
+    }
+
+}

apps/processer/src/test/kotlin/no/iktdev/mediaprocessing/processer/listeners/MockFFmpeg.kt

@@ -19,6 +19,7 @@ class MockFFmpeg(override val listener: Listener, val delayMillis: Long = 500, p
     }
 
     override suspend fun run(argument: MpegArgument) {
+        logFile = File("build/test-log/file.json")
         inputFile = argument.inputFile!!
         listener.onStarted(argument.inputFile!!)
         delay(delayMillis)

apps/processer/src/test/kotlin/no/iktdev/mediaprocessing/processer/listeners/SubtitleTaskListenerTest.kt

@@ -7,6 +7,7 @@ import no.iktdev.eventi.models.store.TaskStatus
 import no.iktdev.eventi.tasks.TaskReporter
 import no.iktdev.eventi.tasks.TaskTypeRegistry
 import no.iktdev.mediaprocessing.ffmpeg.FFmpeg
+import no.iktdev.mediaprocessing.processer.TestUtils
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.events.ProcesserExtractResultEvent
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.tasks.ExtractSubtitleData
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.tasks.ExtractSubtitleTask
@@ -19,7 +20,10 @@ import kotlin.system.measureTimeMillis
 
 class SubtitleTaskListenerTest {
 
-    class TestListener(val delay: Long): SubtitleTaskListener() {
+    class TestListener(val delay: Long): SubtitleTaskListener(
+        fileUtil = TestUtils.getFileUtil(),
+        executableConfig = TestUtils.getExecutableConfig(),
+    ) {
         fun getJob() = currentJob
 
         private var _result: Event? = null

apps/processer/src/test/kotlin/no/iktdev/mediaprocessing/processer/listeners/VideoTaskListenerTest.kt

@@ -9,6 +9,8 @@ import no.iktdev.eventi.tasks.TaskReporter
 import no.iktdev.eventi.tasks.TaskTypeRegistry
 import no.iktdev.mediaprocessing.ffmpeg.FFmpeg
 import no.iktdev.mediaprocessing.processer.CoordinatorClient
+import no.iktdev.mediaprocessing.processer.LocalProgressCache
+import no.iktdev.mediaprocessing.processer.TestUtils
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.events.ProcesserEncodeResultEvent
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.tasks.EncodeData
 import no.iktdev.mediaprocessing.shared.common.event_task_contract.tasks.EncodeTask
@@ -21,7 +23,13 @@ import kotlin.system.measureTimeMillis
 
 class VideoTaskListenerTest {
 
-    class TestListener(val delay: Long, coordinatorClient: CoordinatorClient): VideoTaskListener(coordinatorClient) {
+
+    class TestListener(val delay: Long, coordinatorClient: CoordinatorClient):
+        VideoTaskListener(coordinatorWebClient = coordinatorClient,
+            localProgress = LocalProgressCache(),
+            fileUtil = TestUtils.getFileUtil(),
+            executableConfig = TestUtils.getExecutableConfig(),
+        ) {
         fun getJob() = currentJob
 
         private var _result: Event? = null

apps/processer/src/test/resources/application.yml

@@ -33,12 +33,17 @@ management:
       show-details: always
 
 media:
-  cache: /src/cache
+  cache: build/test-cache
-  outgoing: /src/output
+  outgoing: build/test-output
-  incoming: /src/input
+  incoming: build/test-input
-
-streamit:
-  address: http://streamit.service
 
 executables:
   ffmpeg: ffmpeg
+
+processer:
+  coordinator-url: ${COORDINATOR_URL:http://coordinator:8080}
+  coordinator-ping-on-startup: false
+  allow-overwrite: ${ALLOW_OVERWRITE:false}
+
+directories:
+  logs: build/test-logs

shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/MediaProcessingApplication.kt

@@ -29,7 +29,6 @@ annotation class MediaProcessingApp
 @Configuration
 @EnableConfigurationProperties(
     value = [
-        StreamItConfig::class,
         MediaPaths::class
     ]
 )

shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/event_task_contract/events/ProcesserEncodeResultEvent.kt

@@ -5,6 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.event_task_contract.TaskResultEve
 
 class ProcesserEncodeResultEvent(
     val data: EncodeResult? = null,
+    val logFile: String? = null,
     status: TaskStatus,
     error: String? = null
 ) : TaskResultEvent(status, error) {

shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/rules/TaskLifecycleRules.kt

@@ -11,14 +11,14 @@ object TaskLifecycleRules {
 
     fun isAbandoned(
         consumed: Boolean,
-        createdAt: Instant,
+        persistedAt: Instant,
         lastCheckIn: Instant?
     ): Boolean {
         if (consumed) return false
 
         val cutoff = Instant.now().minus(abandonedAfterMinutes, ChronoUnit.MINUTES)
 
-        val reference = lastCheckIn ?: createdAt
+        val reference = lastCheckIn ?: persistedAt
         return reference.isBefore(cutoff)
     }