[WIP] begin to refactor kafka topology.

src/main/kotlin/KafkaTopology.kt

@@ -21,17 +21,8 @@ package org.memobase
 import com.beust.klaxon.Klaxon
 import org.apache.kafka.streams.KeyValue
 import org.apache.kafka.streams.StreamsBuilder
-import org.apache.kafka.streams.Topology
 import org.apache.kafka.streams.kstream.KStream
-import org.apache.kafka.streams.kstream.Predicate
-import org.memobase.models.ErrorResult
-import org.memobase.models.Formats
-import org.memobase.models.Message
-import org.memobase.models.ParserResult
-import org.memobase.models.ProcessReport
-import org.memobase.models.Report
-import org.memobase.models.ReportMessages
-import org.memobase.models.ReportStatus
+import org.memobase.models.*
 import org.memobase.settings.SettingsLoader
 
 class KafkaTopology(private val settings: SettingsLoader) {
@@ -40,22 +31,19 @@ class KafkaTopology(private val settings: SettingsLoader) {
     private val reportingTopic = settings.outputTopic + "-reporting"
 
     private val klaxon = Klaxon()
+    private val acceptedFormats = listOf(Formats.csv, Formats.xlsx, Formats.tsv, Formats.xls, Formats.ods)
 
     fun prepare(): StreamsBuilder {
         val builder = StreamsBuilder()
 
-        val branchedSource = builder
+        val parsedTable = builder
             .stream<String, String>(settings.inputTopic)
-                .flatMapValues { value -> listOfNotNull(klaxon.parse<Message>(value)) }
-                .branch(
-                        Predicate { _, value -> value.format == Formats.error },
-                        Predicate { _, _ -> true }
-                )
+            .mapValues { value -> klaxon.parse<Message>(value) }
+            .filter { _, value -> acceptedFormats.contains(value?.format) }
+            .mapValues { key, value -> parser.parseTable(key, value!!) }
+
 
-        ignoreFaultyInput(branchedSource[0])
 
-        processValidInput(branchedSource[1]
-                .mapValues { key, value -> parser.parseTable(key, value) })
 
         return builder
     }
@@ -79,28 +67,4 @@ class KafkaTopology(private val settings: SettingsLoader) {
             .map { _, value -> KeyValue(value.processReport.id, value.processReport.toJson()) }
             .to(settings.processReportTopic)
     }
-
-    private fun ignoreFaultyInput(stream: KStream<String, Message>) {
-        stream
-                .mapValues { _ -> ErrorResult.get().toJsonString() }
-                .to(settings.outputTopic)
-
-        stream
-                .mapValues { key, _ ->
-                    Report(
-                            key,
-                            ReportStatus.failure,
-                            ReportMessages.processFailure(key, "of an invalid input message.")
-                    )
-                }
-                .mapValues { value -> value.toJson() }
-                .to("${settings.outputTopic}-reporting")
-
-        stream
-                .mapValues { _ ->
-                    ProcessReport(ReportStatus.failure, 1)
-                }
-                .mapValues { value -> value.toJson() }
-                .to(settings.processReportTopic)
-    }
 }