Update to remove unnecessary code

Remove references to recordSetID

src/main/kotlin/KafkaTopology.kt

@@ -22,10 +22,8 @@ import org.apache.jena.rdf.model.Model
 import org.apache.jena.rdf.model.ModelFactory
 import org.apache.jena.rdf.model.Resource
 import org.apache.jena.rdf.model.ResourceFactory
-import org.apache.jena.rdf.model.impl.StatementImpl
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.kstream.Predicate
-import org.apache.logging.log4j.LogManager
 import org.memobase.rdf.EBUCORE
 import org.memobase.rdf.RDF
 import org.memobase.rdf.RICO
@@ -37,22 +35,13 @@ import settings.HeaderExtractionTransformSupplier
 import settings.HeaderMetadata
 import java.io.StringReader
 import java.io.StringWriter
-import kotlin.system.exitProcess
 
 class KafkaTopology(private val settings: SettingsLoader) {
-    private val log = LogManager.getLogger("MediaLinker")
-
     private val appSettings = settings.appSettings
 
-    private val fileFetcher = LocatorFetcher(
-        settings.sftpSettings,
-        appSettings.getProperty(Constant.sftpBasePathPropertyName),
-        appSettings.getProperty(Constant.recordSetIdPropertyName),
-        appSettings.getProperty(Constant.hasThumbnailsPropertyName)!!.toBoolean()
-    )
     private val sftpClient = SftpClient(settings.sftpSettings)
     private val sftpBasePath = appSettings.getProperty(Constant.sftpBasePathPropertyName)
-    private val reportingTopic = "${settings.processReportTopic}"
+    private val reportingTopic = settings.processReportTopic
 
     fun prepare(): StreamsBuilder {
         val builder = StreamsBuilder()
@@ -189,7 +178,6 @@ class KafkaTopology(private val settings: SettingsLoader) {
                         data.third.message + "\n" + ReportMessages.reportSuccess(digitalObject.uri, link)
                     )
                 )
-                break
             }
         }
         return Triple(

src/main/kotlin/LocatorFetcher.kt

-/*
- * Media Linker
- * 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 org.apache.log4j.LogManager
-import org.memobase.settings.SftpSettings
-import org.memobase.sftp.SftpClient
-
-class LocatorFetcher(sftpSettings: SftpSettings, basePath: String, path: String, private val hasThumbnails: Boolean) {
-    private val log = LogManager.getLogger("LocatorFetcher")
-
-    private val client = SftpClient(sftpSettings)
-    private val pathMedia = "$basePath/$path/${Constant.mediaFolderName}"
-    private val pathThumbnails = "$basePath/$path/${Constant.thumbnailFolderName}"
-
-    fun thumbnailFiles(): List<String> {
-        return if (hasThumbnails) {
-            if (!client.exists(pathThumbnails)) {
-                throw FolderNotFoundException(pathThumbnails)
-            }
-            log.info("Found thumbnail folder on sftp server: $pathThumbnails.")
-            client.listFiles(pathThumbnails)
-        } else {
-            emptyList()
-        }
-    }
-
-    fun mediaFiles(): List<String> {
-        return if (client.exists(pathMedia)) {
-            log.info("Found media folder on sftp server: $pathMedia")
-            client.listFiles(pathMedia)
-        } else {
-            throw FolderNotFoundException(pathMedia)
-        }
-    }
-
-    fun fetchThumbnails() {
-        //TODO: Add rdf:type Instantiation, rico:type Thumbnail & ebucore:locator sftp link
-
-    }
-
-}
\ No newline at end of file

src/test/kotlin/TestKafkaTopology.kt

@@ -83,7 +83,7 @@ class TestKafkaTopology {
         val value = readFile(params.inputFileName)
         val headers = RecordHeaders()
         headers.add(RecordHeader("sessionId", "test-session-id".toByteArray()))
-        headers.add(RecordHeader("recordSetId", (service.settings.appSettings.getProperty(Constant.recordSetIdPropertyName.toString()).toByteArray())))
+        headers.add(RecordHeader("recordSetId", "test-record-set-id".toByteArray()))
         headers.add(RecordHeader("institutionId", "test-institution-id".toByteArray()))
         testDriver.pipeInput(
                 factory.create(service.settings.inputTopic, params.key, value, headers)