pcvolkmer/etl-processor
1
0
Fork 0
mirror of https://github.com/pcvolkmer/etl-processor.git synced 2025-04-19 17:26:51 +00:00
Code Issues Releases Activity

Add request ID to Kafka key

Browse Source
This commit is contained in:
Paul-Christian Volkmer 2023-07-31 10:28:54 +02:00
parent 2d140d94b2
commit aed5f15d2d
4 changed files with 20 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
View File

@ -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}\"}"
}
}

4
src/main/kotlin/dev/dnpm/etl/processor/output/MtbFileSender.kt
View File

@ -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,

5
src/main/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSender.kt
View File

@ -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,

10
src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
View File

@ -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")