From 3e45bf84940b1e94c642bf3a88e94514e5a83c41 Mon Sep 17 00:00:00 2001
From: Paul-Christian Volkmer <volkmer_p@ukw.de>
Date: Thu, 29 Feb 2024 09:19:32 +0100
Subject: [PATCH] feat: implement KafkaInputListener

---
 .../etl/processor/config/AppConfigurationTest.kt |  2 ++
 .../processor/config/AppKafkaConfiguration.kt    |  5 +++--
 .../etl/processor/input/KafkaInputListener.kt    | 16 +++++++++++++---
 3 files changed, 18 insertions(+), 5 deletions(-)

diff --git a/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt b/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
index 46a756b..262aca0 100644
--- a/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
+++ b/src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt
@@ -26,6 +26,7 @@ import dev.dnpm.etl.processor.output.KafkaMtbFileSender
 import dev.dnpm.etl.processor.output.RestMtbFileSender
 import dev.dnpm.etl.processor.pseudonym.AnonymizingGenerator
 import dev.dnpm.etl.processor.pseudonym.GpasPseudonymGenerator
+import dev.dnpm.etl.processor.services.RequestProcessor
 import dev.dnpm.etl.processor.services.TokenRepository
 import dev.dnpm.etl.processor.services.TokenService
 import org.assertj.core.api.Assertions.assertThat
@@ -144,6 +145,7 @@ class AppConfigurationTest {
             "app.kafka.group-id=test"
         ]
     )
+    @MockBean(RequestProcessor::class)
     inner class AppConfigurationUsingKafkaInputTest(private val context: ApplicationContext) {
 
         @Test
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt b/src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt
index 0977527..1ff5e58 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt
@@ -25,6 +25,7 @@ import dev.dnpm.etl.processor.monitoring.ConnectionCheckService
 import dev.dnpm.etl.processor.monitoring.KafkaConnectionCheckService
 import dev.dnpm.etl.processor.output.KafkaMtbFileSender
 import dev.dnpm.etl.processor.output.MtbFileSender
+import dev.dnpm.etl.processor.services.RequestProcessor
 import dev.dnpm.etl.processor.services.kafka.KafkaResponseProcessor
 import org.slf4j.LoggerFactory
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
@@ -97,9 +98,9 @@ class AppKafkaConfiguration {
     @Bean
     @ConditionalOnProperty(value = ["app.kafka.input-topic"])
     fun kafkaInputListener(
-        applicationEventPublisher: ApplicationEventPublisher,
+        requestProcessor: RequestProcessor
     ): KafkaInputListener {
-        return KafkaInputListener(applicationEventPublisher)
+        return KafkaInputListener(requestProcessor)
     }
 
     @Bean
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/input/KafkaInputListener.kt b/src/main/kotlin/dev/dnpm/etl/processor/input/KafkaInputListener.kt
index ee6b56e..f3c0ab4 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/input/KafkaInputListener.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/input/KafkaInputListener.kt
@@ -19,15 +19,25 @@
 
 package dev.dnpm.etl.processor.input
 
+import de.ukw.ccc.bwhc.dto.Consent
 import de.ukw.ccc.bwhc.dto.MtbFile
+import dev.dnpm.etl.processor.services.RequestProcessor
 import org.apache.kafka.clients.consumer.ConsumerRecord
-import org.springframework.context.ApplicationEventPublisher
+import org.slf4j.LoggerFactory
 import org.springframework.kafka.listener.MessageListener
 
 class KafkaInputListener(
-    private val applicationEventPublisher: ApplicationEventPublisher
+    private val requestProcessor: RequestProcessor
 ) : MessageListener<String, MtbFile> {
+    private val logger = LoggerFactory.getLogger(KafkaInputListener::class.java)
+
     override fun onMessage(data: ConsumerRecord<String, MtbFile>) {
-        TODO("Not yet implemented")
+        if (data.value().consent.status == Consent.Status.ACTIVE) {
+            logger.debug("Accepted MTB File for processing")
+            requestProcessor.processMtbFile(data.value())
+        } else {
+            logger.debug("Accepted MTB File and process deletion")
+            requestProcessor.processDeletion(data.value().patient.id)
+        }
     }
 }
\ No newline at end of file