Add request ID to Kafka key

4 files changed, 20 insertions, 14 deletions

diff --git a/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt b/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
index 9867deb..7a88445 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
@@ -20,7 +20,6 @@
 package dev.dnpm.etl.processor.output
 
 import com.fasterxml.jackson.databind.ObjectMapper
-import de.ukw.ccc.bwhc.dto.MtbFile
 import org.slf4j.LoggerFactory
 import org.springframework.kafka.core.KafkaTemplate
 
@@ -31,14 +30,11 @@ class KafkaMtbFileSender(
 
     private val logger = LoggerFactory.getLogger(KafkaMtbFileSender::class.java)
 
-    override fun send(mtbFile: MtbFile): MtbFileSender.Response {
+    override fun send(request: MtbFileSender.Request): MtbFileSender.Response {
         return try {
             val result = kafkaTemplate.sendDefault(
-                String.format(
-                    "{\"pid\": \"%s\", \"eid\": \"%s\"}",
-                    mtbFile.patient.id,
-                    mtbFile.episode.id
-                ), objectMapper.writeValueAsString(mtbFile)
+                header(request),
+                objectMapper.writeValueAsString(request.mtbFile)
             )
             if (result.get() != null) {
                 logger.debug("Sent file via KafkaMtbFileSender")
@@ -53,4 +49,9 @@ class KafkaMtbFileSender(
         }
     }
 
+    private fun header(request: MtbFileSender.Request): String {
+        return "{\"pid\": \"${request.mtbFile.patient.id}\", " +
+                "\"eid\": \"${request.mtbFile.episode.id}\", " +
+                "\"requestId\": \"${request.requestId}\"}"
+    }
 }
\ No newline at end of file
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/output/MtbFileSender.kt b/src/main/kotlin/dev/dnpm/etl/processor/output/MtbFileSender.kt
index d86fd6b..1fdc9e6 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/output/MtbFileSender.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/output/MtbFileSender.kt
@@ -22,10 +22,12 @@ package dev.dnpm.etl.processor.output
 import de.ukw.ccc.bwhc.dto.MtbFile
 
 interface MtbFileSender {
-    fun send(mtbFile: MtbFile): Response
+    fun send(request: Request): Response
 
     data class Response(val status: ResponseStatus, val reason: String = "")
 
+    data class Request(val requestId: String, val mtbFile: MtbFile)
+
     enum class ResponseStatus {
         SUCCESS,
         WARNING,
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSender.kt b/src/main/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSender.kt
index 10f8421..fc3d1c0 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSender.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSender.kt
@@ -19,7 +19,6 @@
 
 package dev.dnpm.etl.processor.output
 
-import de.ukw.ccc.bwhc.dto.MtbFile
 import dev.dnpm.etl.processor.config.RestTargetProperties
 import org.slf4j.LoggerFactory
 import org.springframework.http.HttpEntity
@@ -34,11 +33,11 @@ class RestMtbFileSender(private val restTargetProperties: RestTargetProperties)
 
     private val restTemplate = RestTemplate()
 
-    override fun send(mtbFile: MtbFile): MtbFileSender.Response {
+    override fun send(request: MtbFileSender.Request): MtbFileSender.Response {
         try {
             val headers = HttpHeaders()
             headers.contentType = MediaType.APPLICATION_JSON
-            val entityReq = HttpEntity(mtbFile, headers)
+            val entityReq = HttpEntity(request.mtbFile, headers)
             val response = restTemplate.postForEntity(
                 restTargetProperties.uri!!,
                 entityReq,
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt b/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
index ee13eed..af780bd 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
@@ -35,6 +35,7 @@ import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestBody
 import org.springframework.web.bind.annotation.RestController
 import reactor.core.publisher.Sinks
+import java.util.UUID
 
 @RestController
 class MtbFileController(
@@ -70,8 +71,10 @@ class MtbFileController(
             return ResponseEntity.noContent().build()
         }
 
+        val request = MtbFileSender.Request(UUID.randomUUID().toString(), pseudonymized)
+
         val responses = senders.map {
-            val responseStatus = it.send(pseudonymized)
+            val responseStatus = it.send(request)
             if (responseStatus.status == MtbFileSender.ResponseStatus.SUCCESS || responseStatus.status == MtbFileSender.ResponseStatus.WARNING) {
                 logger.info(
                     "Sent file for Patient '{}' using '{}'",
@@ -100,9 +103,10 @@ class MtbFileController(
 
         requestRepository.save(
             Request(
-                patientId = pseudonymized.patient.id,
+                uuid = request.requestId,
+                patientId = request.mtbFile.patient.id,
                 pid = pid,
-                fingerprint = fingerprint(mtbFile),
+                fingerprint = fingerprint(request.mtbFile),
                 status = requestStatus,
                 report = when (requestStatus) {
                     RequestStatus.ERROR -> Report("Fehler bei der Datenübertragung oder Inhalt nicht verarbeitbar")