diff options
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 |