From 066781edb98770a89d7f08c38874ed7aeaba44ec Mon Sep 17 00:00:00 2001
From: bskjon <bskjon@outlook.com>
Date: Tue, 23 Apr 2024 19:09:45 +0200
Subject: [PATCH] Test

---
 .../input/watcher/InputDirectoryWatcher.kt    | 49 +++++++++++--------
 .../mediaprocessing/processer/Coordinator.kt  | 13 ++++-
 .../processer/services/ClaimsService.kt       |  8 ++-
 .../processer/services/EncodeService.kt       |  1 +
 .../processer/services/ExtractService.kt      |  1 +
 5 files changed, 48 insertions(+), 24 deletions(-)

diff --git a/apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/input/watcher/InputDirectoryWatcher.kt b/apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/input/watcher/InputDirectoryWatcher.kt
index da9cdf1c..78718e53 100644
--- a/apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/input/watcher/InputDirectoryWatcher.kt
+++ b/apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/input/watcher/InputDirectoryWatcher.kt
@@ -41,32 +41,41 @@ class InputDirectoryWatcher(@Autowired var coordinator: Coordinator): FileWatche
     private val logger = KotlinLogging.logger {}
     val watcherChannel = SharedConfig.incomingContent.asWatchChannel()
     val queue = FileWatcherQueue()
-    val io = Coroutines.io()
+    final val io = Coroutines.io()
 
-    init {
-        io.launch {
-            watcherChannel.consumeEach {
-                if (it.file == SharedConfig.incomingContent) {
-                    logger.info { "IO Watcher ${it.kind} on ${it.file.absolutePath}" }
-                } else {
-                    logger.info { "IO Event: ${it.kind}: ${it.file.name}" }
-                }
-                try {
-                    when (it.kind) {
-                        Deleted -> removeFile(it.file)
-                        Initialized -> { /* Do nothing */ }
-                        else -> {
-                            val added = addFile(it.file)
-                            if (!added) {
-                                logger.info { "Ignoring event kind: ${it.kind.name} for file ${it.file.name} as it is not a supported video file" }
-                            }
+    suspend fun startWatcher() {
+        log.info { "Starting Watcher" }
+        watcherChannel.consumeEach {
+            if (it.file == SharedConfig.incomingContent) {
+                logger.info { "IO Watcher ${it.kind} on ${it.file.absolutePath}" }
+            } else {
+                logger.info { "IO Event: ${it.kind}: ${it.file.name}" }
+            }
+            try {
+                when (it.kind) {
+                    Deleted -> removeFile(it.file)
+                    Initialized -> { /* Do nothing */ }
+                    else -> {
+                        val added = addFile(it.file)
+                        if (!added) {
+                            logger.info { "Ignoring event kind: ${it.kind.name} for file ${it.file.name} as it is not a supported video file" }
                         }
                     }
-                } catch (e: Exception) {
-                    e.printStackTrace()
                 }
+            } catch (e: Exception) {
+                e.printStackTrace()
             }
         }
+        log.info { "Reached end of watcherChannel" }
+    }
+
+    final fun runCoroutine() {
+        io.launch { startWatcher() }
+            .invokeOnCompletion { runCoroutine() }
+    }
+
+    init {
+        runCoroutine()
     }
 
     private fun addFile(file: File): Boolean {
diff --git a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/Coordinator.kt b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/Coordinator.kt
index d622f1d5..86844ff9 100644
--- a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/Coordinator.kt
+++ b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/Coordinator.kt
@@ -85,7 +85,7 @@ class Coordinator(): CoordinatorBase<PersistentProcessDataMessage, PersistentEve
         }
     }
 
-    fun readAllAvailableInQueue() {
+    private fun readAllAvailableInQueue() {
         val messages = eventManager.getProcessEventsClaimable()
         io.launch {
             messages.forEach {
@@ -117,7 +117,16 @@ class Coordinator(): CoordinatorBase<PersistentProcessDataMessage, PersistentEve
 
     }
 
-    fun readAllMessagesFor(referenceId: String, eventId: String) {
+    /**
+     * If we get double events at the same time, this would be the case
+     */
+    fun readNextAvailableMessageWithEvent(kafkaEvents: KafkaEvents) {
+        val messages = eventManager.getProcessEventsClaimable().firstOrNull { it.event == kafkaEvents }?.let {
+            readAllMessagesFor(referenceId = it.referenceId, eventId = it.eventId)
+        }
+    }
+
+    private fun readAllMessagesFor(referenceId: String, eventId: String) {
         val messages = eventManager.getProcessEventsClaimable()
         createTasksBasedOnEventsAndPersistence(referenceId, eventId, messages)
     }
diff --git a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ClaimsService.kt b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ClaimsService.kt
index 8cc58b78..8c3a8040 100644
--- a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ClaimsService.kt
+++ b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ClaimsService.kt
@@ -16,6 +16,10 @@ class ClaimsService() {
     @Autowired
     lateinit var coordinator: Coordinator
 
+
+    /**
+     * If this serivce calls for readback all on the coordinator, it will cause a full on crash, as it
+     */
     @Scheduled(fixedDelay = (300_000))
     fun validateClaims() {
         val expiredClaims = eventManager.getProcessEventsWithExpiredClaim()
@@ -31,9 +35,9 @@ class ClaimsService() {
             }
             it
         }
-        released.forEach {
+        /*released.forEach {
             log.info { "Sending released ${it.referenceId} ${it.event} into queue" }
             coordinator.readAllMessagesFor(it.referenceId, it.eventId)
-        }
+        }*/
     }
 }
\ No newline at end of file
diff --git a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/EncodeService.kt b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/EncodeService.kt
index f21c6cde..0742b6d0 100644
--- a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/EncodeService.kt
+++ b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/EncodeService.kt
@@ -209,6 +209,7 @@ class EncodeService(@Autowired override var coordinator: Coordinator, @Autowired
 
     fun clearWorker() {
         this.runner = null
+        coordinator.readNextAvailableMessageWithEvent(KafkaEvents.EventWorkEncodeCreated)
     }
 
     @PreDestroy
diff --git a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ExtractService.kt b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ExtractService.kt
index af0014dd..24d55432 100644
--- a/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ExtractService.kt
+++ b/apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ExtractService.kt
@@ -188,6 +188,7 @@ class ExtractService(@Autowired override var coordinator: Coordinator, @Autowire
 
     fun clearWorker() {
         this.runner = null
+        coordinator.readNextAvailableMessageWithEvent(KafkaEvents.EventWorkExtractCreated)
     }
 
     @PreDestroy