-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Read from opprett sykmelding topic (#234)
* Added OpprettSykmeldingConsumer and tried to create sykmelding Co-authored-by: Andreas <danduras@gmail.com> Co-authored-by: Helene Arnesen <helene.arnesen@nav.no> * updated logging Co-authored-by: Andreas <danduras@gmail.com> Co-authored-by: Helene Arnesen <helene.arnesen@nav.no> * more log Co-authored-by: Joakim Taule Kartveit <joakimkartveit@gmail.com> * Use Dispatchers.IO --------- Co-authored-by: Jørn-Are Flaten <ja.flaten91@gmail.com> Co-authored-by: Andreas <danduras@gmail.com> Co-authored-by: Helene Arnesen <helene.arnesen@nav.no>
- Loading branch information
4 people
committed
Nov 22, 2023
1 parent
1797241
commit f9ded01
Showing
8 changed files
with
192 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
20 changes: 20 additions & 0 deletions
20
src/main/kotlin/no/nav/syfo/opprettsykmelding/OpprettSykmeldingDeserializer.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| package no.nav.syfo.opprettsykmelding | ||
|
|
||
| import com.fasterxml.jackson.databind.DeserializationFeature | ||
| import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule | ||
| import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper | ||
| import com.fasterxml.jackson.module.kotlin.readValue | ||
| import no.nav.syfo.opprettsykmelding.model.OpprettSykmeldingRecord | ||
| import org.apache.kafka.common.serialization.Deserializer | ||
|
|
||
| class OpprettSykmeldingDeserializer : Deserializer<OpprettSykmeldingRecord> { | ||
| private val objectMapper = | ||
| jacksonObjectMapper().apply { | ||
| registerModule(JavaTimeModule()) | ||
| configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) | ||
| } | ||
|
|
||
| override fun deserialize(topic: String, data: ByteArray): OpprettSykmeldingRecord { | ||
| return objectMapper.readValue(data) | ||
| } | ||
| } |
126 changes: 126 additions & 0 deletions
126
src/main/kotlin/no/nav/syfo/opprettsykmelding/OpprettSykmeldingService.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| package no.nav.syfo.opprettsykmelding | ||
|
|
||
| import java.time.Duration | ||
| import java.util.* | ||
| import kotlinx.coroutines.DelicateCoroutinesApi | ||
| import net.logstash.logback.argument.StructuredArguments.fields | ||
| import no.nav.syfo.Environment | ||
| import no.nav.syfo.application.ApplicationState | ||
| import no.nav.syfo.client.SafJournalpostClient | ||
| import no.nav.syfo.createListener | ||
| import no.nav.syfo.kafka.aiven.KafkaUtils | ||
| import no.nav.syfo.kafka.toConsumerConfig | ||
| import no.nav.syfo.opprettsykmelding.model.OpprettSykmeldingRecord | ||
| import no.nav.syfo.pdl.service.PdlPersonService | ||
| import no.nav.syfo.securelog | ||
| import no.nav.syfo.service.SykmeldingService | ||
| import no.nav.syfo.util.LoggingMeta | ||
| import org.apache.kafka.clients.consumer.KafkaConsumer | ||
| import org.apache.kafka.common.serialization.StringDeserializer | ||
| import org.slf4j.LoggerFactory | ||
|
|
||
| @OptIn(DelicateCoroutinesApi::class) | ||
| fun startOpprettSykmeldingConsumer( | ||
| environment: Environment, | ||
| applicationState: ApplicationState, | ||
| sykmeldingService: SykmeldingService, | ||
| safJournalpostClient: SafJournalpostClient, | ||
| pdlPersonService: PdlPersonService | ||
| ) { | ||
|
|
||
| val consumer = "opprett-sykmelding-consumer" | ||
| val consumerProperties = | ||
| KafkaUtils.getAivenKafkaConfig(consumer) | ||
| .toConsumerConfig( | ||
| groupId = consumer, | ||
| valueDeserializer = OpprettSykmeldingDeserializer::class | ||
| ) | ||
|
|
||
| val kafkaConsumer = | ||
| KafkaConsumer(consumerProperties, StringDeserializer(), OpprettSykmeldingDeserializer()) | ||
|
|
||
| OpprettSykmeldingService( | ||
| kafkaConsumer = kafkaConsumer, | ||
| sykmeldingService = sykmeldingService, | ||
| env = environment, | ||
| applicationState = applicationState, | ||
| safJournalpostClient = safJournalpostClient, | ||
| pdlPersonService = pdlPersonService, | ||
| ) | ||
| .start() | ||
| } | ||
|
|
||
| class OpprettSykmeldingService( | ||
| private val kafkaConsumer: KafkaConsumer<String, OpprettSykmeldingRecord>, | ||
| private val sykmeldingService: SykmeldingService, | ||
| private val env: Environment, | ||
| private val applicationState: ApplicationState, | ||
| private val safJournalpostClient: SafJournalpostClient, | ||
| private val pdlPersonService: PdlPersonService | ||
| ) { | ||
| companion object { | ||
| private val log = LoggerFactory.getLogger(OpprettSykmeldingService::class.java) | ||
| } | ||
|
|
||
| private val journalPostQuery = | ||
| SafJournalpostClient::class | ||
| .java | ||
| .getResource("/graphql/findJournalpost.graphql")!! | ||
| .readText() | ||
| .replace(Regex("[\n\t]"), "") | ||
|
|
||
| @DelicateCoroutinesApi | ||
| fun start() { | ||
| createListener(applicationState) { consumeTopic() } | ||
| } | ||
|
|
||
| suspend fun consumeTopic() { | ||
| kafkaConsumer.subscribe(listOf(env.opprettSykmeldingTopic)) | ||
| while (applicationState.ready) { | ||
| kafkaConsumer.poll(Duration.ofMillis(1000)).forEach { consumerRecord -> | ||
| val opprettSykmeldingRecord = consumerRecord.value() | ||
| val sykmeldingId = UUID.randomUUID().toString() | ||
| val loggingMeta = | ||
| LoggingMeta( | ||
| sykmeldingId = sykmeldingId, | ||
| journalpostId = opprettSykmeldingRecord.data.journalpostId, | ||
| hendelsesId = opprettSykmeldingRecord.data.journalpostId, | ||
| ) | ||
| val journalpostId = opprettSykmeldingRecord.data.journalpostId | ||
| log.info("received opprett sykmelding for $journalpostId {}", fields(loggingMeta)) | ||
| securelog.info( | ||
| "received opprett sykmelding message $opprettSykmeldingRecord {}", | ||
| fields(loggingMeta) | ||
| ) | ||
| val journalpostMetadata = | ||
| safJournalpostClient.getJournalpostMetadata( | ||
| journalpostId, | ||
| journalPostQuery, | ||
| loggingMeta, | ||
| ) | ||
| ?: throw IllegalStateException( | ||
| "Unable to find journalpost with id $journalpostId", | ||
| ) | ||
|
|
||
| val fnrEllerAktorId = | ||
| when (journalpostMetadata.bruker.type) { | ||
| "ORGNR" -> throw IllegalStateException("Bruker id is ORGNR") | ||
| else -> journalpostMetadata.bruker.id | ||
| } | ||
| ?: throw IllegalStateException("Bruker id is null") | ||
|
|
||
| val pasient = pdlPersonService.getPdlPerson(fnrEllerAktorId, loggingMeta) | ||
| sykmeldingService.behandleSykmelding( | ||
| journalpostId = journalpostId, | ||
| pasient = pasient, | ||
| datoOpprettet = journalpostMetadata.datoOpprettet, | ||
| dokumentInfoIdPdf = journalpostMetadata.dokumentInfoIdPdf, | ||
| dokumentInfoId = journalpostMetadata.dokumentInfoId, | ||
| temaEndret = false, | ||
| loggingMeta = loggingMeta, | ||
| sykmeldingId = sykmeldingId, | ||
| ) | ||
| } | ||
| } | ||
| } | ||
| } |
16 changes: 16 additions & 0 deletions
16
src/main/kotlin/no/nav/syfo/opprettsykmelding/model/OpprettSykmeldingRecord.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package no.nav.syfo.opprettsykmelding.model | ||
|
|
||
| data class Metadata( | ||
| val type: String = "journalpost", | ||
| val source: String = "syk-dig", | ||
| ) | ||
|
|
||
| data class JournalpostMetadata( | ||
| val journalpostId: String, | ||
| val tema: String, | ||
| ) | ||
|
|
||
| data class OpprettSykmeldingRecord( | ||
| val metadata: Metadata, | ||
| val data: JournalpostMetadata, | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters