From e560824b26cc90a1371218e3adbd86c0570167e0 Mon Sep 17 00:00:00 2001
From: Brage <bskjon@outlook.com>
Date: Tue, 25 Jul 2023 00:46:01 +0200
Subject: [PATCH] Update

---
 .../streamit/content/convert/ConvertRunner.kt     | 15 ++++++++++++---
 .../content/convert/kafka/SubtitleConsumer.kt     |  2 ++
 .../reader/collector/ConvertedSubtitleConsumer.kt |  2 ++
 3 files changed, 16 insertions(+), 3 deletions(-)

diff --git a/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/ConvertRunner.kt b/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/ConvertRunner.kt
index 042c4e12..eddf2109 100644
--- a/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/ConvertRunner.kt
+++ b/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/ConvertRunner.kt
@@ -1,5 +1,7 @@
 package no.iktdev.streamit.content.convert
 
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 import no.iktdev.library.subtitle.Syncro
 import no.iktdev.library.subtitle.export.Export
 import no.iktdev.library.subtitle.reader.BaseReader
@@ -17,10 +19,15 @@ class ConvertRunner(val referenceId: String, val listener: IConvertListener) {
         val reader = getReade(subtitleInfo.inputFile)
         val dialogs = reader?.read()
         if (dialogs.isNullOrEmpty()) {
-            listener.onError(referenceId, subtitleInfo, "Dialogs read from file is null or empty!")
+            withContext(Dispatchers.Default) {
+                listener.onError(referenceId, subtitleInfo, "Dialogs read from file is null or empty!")
+            }
             return
         }
-        listener.onStarted(referenceId, subtitleInfo)
+
+        withContext(Dispatchers.Default) {
+            listener.onStarted(referenceId, subtitleInfo)
+        }
 
         val syncedDialogs = Syncro().sync(dialogs)
 
@@ -32,7 +39,9 @@ class ConvertRunner(val referenceId: String, val listener: IConvertListener) {
                 language = subtitleInfo.language,
                 outFile = it.absolutePath
             )
-             listener.onEnded(referenceId, subtitleInfo, work = item)
+             withContext(Dispatchers.Default) {
+                 listener.onEnded(referenceId, subtitleInfo, work = item)
+             }
         }
 
     }
diff --git a/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/kafka/SubtitleConsumer.kt b/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/kafka/SubtitleConsumer.kt
index 6bdd7055..290b91e9 100644
--- a/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/kafka/SubtitleConsumer.kt
+++ b/Convert/src/main/kotlin/no/iktdev/streamit/content/convert/kafka/SubtitleConsumer.kt
@@ -34,11 +34,13 @@ class SubtitleConsumer: DefaultKafkaReader("convertHandlerSubtitle"), IConvertLi
             val workResult = data.value().dataAs(ExtractWork::class.java)
 
             if (workResult?.produceConvertEvent == true) {
+                logger.info { "Using ${data.value().referenceId} ${workResult.outFile} as it is a convert candidate" }
                 val convertWork = SubtitleInfo(
                     inputFile = File(workResult.outFile),
                     collection = workResult.collection,
                     language = workResult.language,
                 )
+                produceMessage(KafkaEvents.EVENT_CONVERTER_STARTED_SUBTITLE_FILE, Message(referenceId = referenceId, Status(statusType = StatusType.PENDING)), convertWork)
                 Coroutines.io().launch {
                     ConvertRunner(referenceId, this@SubtitleConsumer).readAndConvert(convertWork)
                 }
diff --git a/Reader/src/main/kotlin/no/iktdev/streamit/content/reader/collector/ConvertedSubtitleConsumer.kt b/Reader/src/main/kotlin/no/iktdev/streamit/content/reader/collector/ConvertedSubtitleConsumer.kt
index 7eb5181f..b526130e 100644
--- a/Reader/src/main/kotlin/no/iktdev/streamit/content/reader/collector/ConvertedSubtitleConsumer.kt
+++ b/Reader/src/main/kotlin/no/iktdev/streamit/content/reader/collector/ConvertedSubtitleConsumer.kt
@@ -10,8 +10,10 @@ import no.iktdev.streamit.library.kafka.dto.Message
 import no.iktdev.streamit.library.kafka.listener.SimpleMessageListener
 import no.iktdev.streamit.library.kafka.listener.deserializer.IMessageDataDeserialization
 import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.springframework.stereotype.Service
 import java.io.File
 
+@Service
 class ConvertedSubtitleConsumer : DefaultKafkaReader("collectorConsumerConvertedSubtitle") {
 
     private val listener = object: SimpleMessageListener(