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

---
 .../common/deserializers/ConvertWorkDeserializer.kt | 13 +++++++++++++
 .../common/deserializers/DeserializerRegistry.kt    |  1 +
 .../reader/collector/ConvertedSubtitleConsumer.kt   |  5 +++--
 3 files changed, 17 insertions(+), 2 deletions(-)
 create mode 100644 CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/ConvertWorkDeserializer.kt

diff --git a/CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/ConvertWorkDeserializer.kt b/CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/ConvertWorkDeserializer.kt
new file mode 100644
index 00000000..183f0447
--- /dev/null
+++ b/CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/ConvertWorkDeserializer.kt
@@ -0,0 +1,13 @@
+package no.iktdev.streamit.content.common.deserializers
+
+import no.iktdev.streamit.content.common.dto.reader.work.ConvertWork
+import no.iktdev.streamit.content.common.dto.reader.work.EncodeWork
+import no.iktdev.streamit.content.common.dto.reader.work.ExtractWork
+import no.iktdev.streamit.library.kafka.dto.Message
+import no.iktdev.streamit.library.kafka.listener.deserializer.IMessageDataDeserialization
+
+class ConvertWorkDeserializer: IMessageDataDeserialization<ConvertWork> {
+    override fun deserialize(incomingMessage: Message): ConvertWork? {
+        return incomingMessage.dataAs(ConvertWork::class.java)
+    }
+}
\ No newline at end of file
diff --git a/CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/DeserializerRegistry.kt b/CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/DeserializerRegistry.kt
index 59a3c706..0b1f6792 100644
--- a/CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/DeserializerRegistry.kt
+++ b/CommonCode/src/main/java/no/iktdev/streamit/content/common/deserializers/DeserializerRegistry.kt
@@ -16,6 +16,7 @@ class DeserializerRegistry {
             KafkaEvents.EVENT_ENCODER_ENDED_VIDEO_FILE to EncodeWorkDeserializer(),
             KafkaEvents.EVENT_READER_ENCODE_GENERATED_SUBTITLE to ExtractWorkDeserializer(),
             KafkaEvents.EVENT_ENCODER_ENDED_SUBTITLE_FILE to ExtractWorkDeserializer(),
+            KafkaEvents.EVENT_CONVERTER_ENDED_SUBTITLE_FILE to ConvertWorkDeserializer()
 
         )
         fun getRegistry(): Map<KafkaEvents, IMessageDataDeserialization<*>> = _registry.toMap()
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 b526130e..a9d0cf3e 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
@@ -3,6 +3,7 @@ package no.iktdev.streamit.content.reader.collector
 import no.iktdev.streamit.content.common.CommonConfig
 import no.iktdev.streamit.content.common.DefaultKafkaReader
 import no.iktdev.streamit.content.common.deserializers.DeserializerRegistry
+import no.iktdev.streamit.content.common.dto.reader.work.ConvertWork
 import no.iktdev.streamit.content.common.dto.reader.work.ExtractWork
 import no.iktdev.streamit.library.db.query.SubtitleQuery
 import no.iktdev.streamit.library.kafka.KafkaEvents
@@ -22,7 +23,7 @@ class ConvertedSubtitleConsumer : DefaultKafkaReader("collectorConsumerConverted
         accepts = listOf(KafkaEvents.EVENT_CONVERTER_ENDED_SUBTITLE_FILE.event)
     ) {
         override fun onMessageReceived(data: ConsumerRecord<String, Message>) {
-            val workResult = data.value().dataAs(ExtractWork::class.java)
+            val workResult = data.value().dataAs(ConvertWork::class.java)
             if (!data.value().isSuccessful() || workResult == null) {
                 return
             }
@@ -42,6 +43,6 @@ class ConvertedSubtitleConsumer : DefaultKafkaReader("collectorConsumerConverted
     }
 
     override fun loadDeserializers(): Map<String, IMessageDataDeserialization<*>> {
-        return DeserializerRegistry.getEventToDeserializer(KafkaEvents.EVENT_ENCODER_ENDED_SUBTITLE_FILE)
+        return DeserializerRegistry.getEventToDeserializer(KafkaEvents.EVENT_CONVERTER_ENDED_SUBTITLE_FILE)
     }
 }
\ No newline at end of file