KafkaTopology.kt

/*
 * XML Data Import Service
 * Copyright (C) 2020 Memoriav
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

package org.memobase

import ch.memobase.kafka.utils.ConfigJoiner
import ch.memobase.kafka.utils.models.ImportService
import ch.memobase.kafka.utils.models.JoinedValues
import ch.memobase.reporting.Report
import ch.memobase.reporting.ReportStatus
import ch.memobase.settings.HeaderExtractionTransformSupplier
import ch.memobase.settings.SettingsLoader
import ch.memobase.sftp.SftpClient
import java.io.File
import net.sf.saxon.s9api.SaxonApiException
import org.apache.kafka.common.serialization.Serdes
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.apache.logging.log4j.LogManager
import org.memobase.models.Content
import org.memobase.models.Formats
import org.memobase.models.Input
import org.memobase.models.Message
import org.memobase.utils.MissingIdentifierException
import org.memobase.utils.SerdeMessage
import org.memobase.utils.XsltException
import org.memobase.xml.XMLTransformer


class KafkaTopology(private val settings: SettingsLoader) {
    private val log = LogManager.getLogger("XmlDataTopology")
    private val sftpClient: SftpClient = SftpClient(settings.sftpSettings)
    private val xmlTransformer = XMLTransformer()
    private val configJoiner = ConfigJoiner<Message, ByteArray>(
        ImportService.Transform,
        SerdeMessage(),
        Serdes.ByteArray(),
        xmlTransformer::xsltFunction
    )
    private val reportingTopic = settings.processReportTopic

    fun build(): Topology {
        val builder = StreamsBuilder()

        val configStream = builder.stream<String, String>("import-process-config")
            .map { key, value -> KeyValue(key.toByteArray(), value.toByteArray()) }


        val dataStream = builder
            .stream<String, String>(settings.inputTopic)
            .flatMapValues { _, value -> parseMessage(value) }
            .filter { _, value ->
                value.format == Formats.xml
            }

        val joinedStream = configJoiner.join(dataStream, configStream)

        val configNullBranch = joinedStream
            .branch(
                Predicate { _, value -> value.right == null },
                Predicate { _, _ -> true }
            )

        configNullBranch[0]
            .transformValues(HeaderExtractionTransformSupplier<JoinedValues<Message, ByteArray>>())
            .mapValues { key, value ->
                Report(
                    key,
                    ReportStatus.fatal,
                    "Could not find a matching xslt configuration for record set ${value.second.recordSetId}.",
                    Service.name
                ).toJson()
            }
            .to(reportingTopic)

        val saxHandlerStream = configNullBranch[1]
            .mapValues { value ->
                Input(value.left, value.right)
            }
            .transformValues(HeaderExtractionTransformSupplier<Input>())
            .mapValues { value ->
                Content(
                    value.first.message,
                    value.second,
                    value.first.xsltData,
                    sftpClient.open(File(value.first.message.path))
                )
            }
            .map { key, value ->
                value.inputStream.use {
                    try {
                        val output =
                            xmlTransformer.applyXSLT(key, value.headerMetadata, value.inputStream, value.xsltData)
                        KeyValue(
                            output.first,
                            Pair(output.second.output.toString(), output.second.getReport())
                        )
                    } catch (ex: XsltException) {
                        log.error(ex.message)
                        KeyValue(
                            key, Pair(
                                null, Report(
                                    key,
                                    ReportStatus.fatal,
                                    ex.localizedMessage,
                                    Service.name
                                )
                            )
                        )
                    } catch (ex: MissingIdentifierException) {
                        log.error(ex.message)
                        KeyValue(
                            key, Pair(
                                null, Report(
                                    key,
                                    ReportStatus.fatal,
                                    ex.localizedMessage,
                                    Service.name
                                )
                            )
                        )
                    } catch (ex: SaxonApiException) {
                        log.error(ex.message)
                        KeyValue(
                            key, Pair(
                                null, Report(
                                    key,
                                    ReportStatus.fatal,
                                    ex.localizedMessage,
                                    Service.name
                                )
                            )
                        )
                    }
                }
            }
        streamOutput(saxHandlerStream)
        return builder.build()
    }

    private fun streamOutput(stream: KStream<String, Pair<String?, Report>>) {
        stream
            .filter { _, value -> value != null }
            .mapValues { value -> value.first }
            .to(settings.outputTopic)

        stream
            .mapValues { value -> value.second.toJson() }
            .to(reportingTopic)
    }

    private fun parseMessage(value: String): List<Message> {
        return Message.fromJson(value)
    }
}