implement csv validation

build.gradle

@@ -48,6 +48,11 @@ dependencies {
     implementation 'com.hierynomus:sshj:0.27.0'
     // YAML Parser
     implementation 'org.snakeyaml:snakeyaml-engine:2.1'
+    // CSV Reader
+    implementation("com.github.doyaaaaaken:kotlin-csv-jvm:0.7.3")
+    // JSON Parser
+    implementation 'com.beust:klaxon:5.2'
+
 
     compile 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
     compile "org.jetbrains.kotlin:kotlin-script-runtime:1.3.71"

src/main/kotlin/App.kt

@@ -18,6 +18,8 @@
 
 package org.memobase
 
+import com.github.doyaaaaaken.kotlincsv.dsl.csvReader
+import com.github.doyaaaaaken.kotlincsv.util.MalformedCSVException
 import org.apache.logging.log4j.LogManager
 import java.io.File
 import kotlin.system.exitProcess
@@ -30,19 +32,20 @@ class App {
                 val settings = SettingsLoader()
                 val producer = Producer(settings.kafkaProducerProperties, settings.topic)
                 val sftp = SftpClient(settings.sftpSettings)
+                val validator = FileValidation(sftp)
 
 
-
-                val files = sftp.listFiles(settings.appSettings["directory"].orEmpty())
+                val files = sftp.listFiles(settings.appSettings["directory"].orEmpty()).map { File(it) }
                 log.error(files)
 
                 for (file in files) {
-                    if (File(file).extension == "csv") {
-                        log.error(sftp.open(file).length())
+                    val report = validator.validate(file)
+                    when (report.status) {
+                        "valid" -> producer.sendMessage(file.name, file.path)
+                        else -> producer.sendMessage(file.name, "ERROR")
                     }
+                    producer.sendReport(report)
                 }
-
-
                 producer.close()
                 sftp.close()
             } catch (ex: Exception) {

src/main/kotlin/FileValidation.kt

+/*
+ * sftp-reader
+ * 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 com.github.doyaaaaaken.kotlincsv.dsl.csvReader
+import com.github.doyaaaaaken.kotlincsv.util.MalformedCSVException
+import org.apache.logging.log4j.LogManager
+import java.io.File
+
+class FileValidation(val sftp: SftpClient) {
+    private val log = LogManager.getLogger("FileValidation")
+
+
+
+    fun validate(file: File): Report {
+        val remoteFile = sftp.open(file)
+        val stream = remoteFile.RemoteFileInputStream()
+        try {
+            csvReader().readAll(stream)
+        } catch (ex: MalformedCSVException) {
+            remoteFile.close()
+            return Report(id = file.name, status = "invalid", path = file.path, error = ex.localizedMessage)
+        }
+        remoteFile.close()
+        return Report(id = file.name, status = "valid", path = file.path, error = "")
+    }
+
+
+
+}
\ No newline at end of file

src/main/kotlin/Producer.kt

@@ -17,6 +17,7 @@
  */
 package org.memobase
 
+import com.beust.klaxon.Klaxon
 import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.ProducerRecord
 import java.util.*
@@ -26,13 +27,13 @@ class Producer(props: Properties, private val topic: String) {
     private val instance = KafkaProducer<String, String>(props)
     private val reportingTopic = "$topic-reporting"
 
-    fun send(key: String, message: String) {
+    fun sendMessage(key: String, message: String) {
         instance.send(ProducerRecord(topic, key, message))
     }
 
-    fun sendResults(key: String, message: String) {
+    fun sendReport(report: Report) {
         // TODO: Implement Report message!
-        instance.send(ProducerRecord(reportingTopic, key))
+        instance.send(ProducerRecord(reportingTopic, report.id, Klaxon().toJsonString(report)))
     }
 
 

src/main/kotlin/Report.kt

+/*
+ * sftp-reader
+ * 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
+
+data class Report(
+    val id: String,
+    val status: String,
+    val path: String,
+    val error: String
+)
\ No newline at end of file

src/main/kotlin/SftpClient.kt

@@ -22,6 +22,7 @@ import net.schmizz.sshj.SSHClient
 import net.schmizz.sshj.sftp.*
 import net.schmizz.sshj.userauth.UserAuthException
 import org.apache.logging.log4j.LogManager
+import java.io.File
 import kotlin.system.exitProcess
 
 class SftpClient(sftpSettings: Map<String, String>) {
@@ -52,8 +53,8 @@ class SftpClient(sftpSettings: Map<String, String>) {
         return ls(path).filter { fileAttributes(it).mode.type == FileMode.Type.REGULAR }
     }
 
-    fun open(path: String): RemoteFile {
-        return instance.open(path, setOf(OpenMode.READ))
+    fun open(file: File): RemoteFile {
+        return instance.open(file.path, setOf(OpenMode.READ))
     }