From c6b675636e1f6b86353da00bff99f629fa1b93f3 Mon Sep 17 00:00:00 2001 From: Anders Rognstad Date: Tue, 27 Feb 2024 15:49:40 +0100 Subject: [PATCH 1/2] IS-2091: Implement ArbeidstakervarselProducer --- .../esyfovarsel/ArbeidstakervarselProducer.kt | 18 +++++++++-- .../KafkaArbeidstakervarselSerializer.kt | 11 +++++++ .../esyfovarsel/dto/EsyfovarselHendelse.kt | 30 +++++++++++++++++++ .../application/service/VarselServiceSpek.kt | 17 +++++++++-- 4 files changed, 70 insertions(+), 6 deletions(-) create mode 100644 src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/KafkaArbeidstakervarselSerializer.kt create mode 100644 src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/dto/EsyfovarselHendelse.kt diff --git a/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakervarselProducer.kt b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakervarselProducer.kt index 47b526dd..3e8eba08 100644 --- a/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakervarselProducer.kt +++ b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakervarselProducer.kt @@ -3,21 +3,33 @@ package no.nav.syfo.infrastructure.kafka.esyfovarsel import no.nav.syfo.application.IVarselProducer import no.nav.syfo.domain.PersonIdent import no.nav.syfo.domain.Varsel +import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.* import java.util.* import org.apache.kafka.clients.producer.KafkaProducer import org.apache.kafka.clients.producer.ProducerRecord import org.slf4j.LoggerFactory -// TODO: Add esyfovarsel-hendelse as kafkaproducer value -class ArbeidstakervarselProducer(private val kafkaArbeidstakervarselProducer: KafkaProducer) : IVarselProducer { +class ArbeidstakervarselProducer(private val kafkaArbeidstakervarselProducer: KafkaProducer) : IVarselProducer { override fun sendArbeidstakerVarsel(personIdent: PersonIdent, varsel: Varsel) { + val varselHendelse = ArbeidstakerHendelse( + type = HendelseType.SM_ARBEIDSUFORHET_FORHANDSVARSEL, + arbeidstakerFnr = personIdent.value, + data = VarselData( + journalpost = VarselDataJournalpost( + uuid = varsel.uuid.toString(), + id = varsel.journalpostId, + ), + ), + orgnummer = null, + ) + try { kafkaArbeidstakervarselProducer.send( ProducerRecord( ESYFOVARSEL_TOPIC, UUID.randomUUID().toString(), - "", + varselHendelse, ) ).get() } catch (e: Exception) { diff --git a/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/KafkaArbeidstakervarselSerializer.kt b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/KafkaArbeidstakervarselSerializer.kt new file mode 100644 index 00000000..f5d53806 --- /dev/null +++ b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/KafkaArbeidstakervarselSerializer.kt @@ -0,0 +1,11 @@ +package no.nav.syfo.infrastructure.kafka.esyfovarsel + +import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.EsyfovarselHendelse +import no.nav.syfo.util.configuredJacksonMapper +import org.apache.kafka.common.serialization.Serializer + +class KafkaArbeidstakervarselSerializer : Serializer { + private val mapper = configuredJacksonMapper() + override fun serialize(topic: String?, data: EsyfovarselHendelse?): ByteArray = + mapper.writeValueAsBytes(data) +} diff --git a/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/dto/EsyfovarselHendelse.kt b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/dto/EsyfovarselHendelse.kt new file mode 100644 index 00000000..49ac3bbc --- /dev/null +++ b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/dto/EsyfovarselHendelse.kt @@ -0,0 +1,30 @@ +package no.nav.syfo.infrastructure.kafka.esyfovarsel.dto + +import com.fasterxml.jackson.annotation.JsonTypeInfo +import java.io.Serializable + +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME) +sealed interface EsyfovarselHendelse : Serializable { + val type: HendelseType + var data: Any? +} + +data class ArbeidstakerHendelse( + override val type: HendelseType, + override var data: Any?, + val arbeidstakerFnr: String, + val orgnummer: String? +) : EsyfovarselHendelse + +data class VarselData( + val journalpost: VarselDataJournalpost? = null, +) + +data class VarselDataJournalpost( + val uuid: String, + val id: String? +) + +enum class HendelseType { + SM_ARBEIDSUFORHET_FORHANDSVARSEL, +} diff --git a/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt b/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt index 638ffae8..c53d6d87 100644 --- a/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt +++ b/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt @@ -9,10 +9,15 @@ import no.nav.syfo.infrastructure.database.repository.VarselRepository import no.nav.syfo.infrastructure.database.repository.VurderingRepository import no.nav.syfo.infrastructure.database.dropData import no.nav.syfo.infrastructure.kafka.esyfovarsel.ArbeidstakervarselProducer +import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.ArbeidstakerHendelse +import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.EsyfovarselHendelse +import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.HendelseType +import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.VarselData import org.amshove.kluent.shouldBeEmpty import org.amshove.kluent.shouldBeEqualTo import org.amshove.kluent.shouldNotBeNull import org.apache.kafka.clients.producer.KafkaProducer +import org.apache.kafka.clients.producer.ProducerRecord import org.apache.kafka.clients.producer.RecordMetadata import org.spekframework.spek2.Spek import org.spekframework.spek2.style.specification.describe @@ -27,7 +32,7 @@ class VarselServiceSpek : Spek({ val varselRepository = VarselRepository(database = database) val vurderingRepository = VurderingRepository(database = database) - val kafkaProducer = mockk>() + val kafkaProducer = mockk>() val varselProducer = ArbeidstakervarselProducer(kafkaArbeidstakervarselProducer = kafkaProducer) val varselService = VarselService(varselRepository = varselRepository, varselProducer = varselProducer) @@ -60,7 +65,8 @@ class VarselServiceSpek : Spek({ failed.size shouldBeEqualTo 0 success.size shouldBeEqualTo 1 - verify(exactly = 1) { kafkaProducer.send(any()) } + val producerRecordSlot = slot>() + verify(exactly = 1) { kafkaProducer.send(capture(producerRecordSlot)) } val publishedVarsel = success.first().getOrThrow() publishedVarsel.uuid.shouldBeEqualTo(unpublishedVarsel.uuid) @@ -69,7 +75,12 @@ class VarselServiceSpek : Spek({ varselRepository.getUnpublishedVarsler().shouldBeEmpty() - // TODO: Test kafkaproducer record value + val esyfovarselHendelse = producerRecordSlot.captured.value() as ArbeidstakerHendelse + esyfovarselHendelse.type.shouldBeEqualTo(HendelseType.SM_ARBEIDSUFORHET_FORHANDSVARSEL) + esyfovarselHendelse.arbeidstakerFnr.shouldBeEqualTo(UserConstants.ARBEIDSTAKER_PERSONIDENT.value) + val varselData = esyfovarselHendelse.data as VarselData + varselData.journalpost?.uuid.shouldBeEqualTo(publishedVarsel.uuid.toString()) + varselData.journalpost?.id.shouldBeEqualTo(publishedVarsel.journalpostId) } it("publishes nothing when no unpublished varsel") { From 4a7ee413b15552eb610743f7086c7f6334db26f0 Mon Sep 17 00:00:00 2001 From: Anders Rognstad Date: Wed, 28 Feb 2024 08:55:31 +0100 Subject: [PATCH 2/2] IS-2091: Rename producer implementation --- .../kotlin/no/nav/syfo/application/IVarselProducer.kt | 2 +- .../no/nav/syfo/application/service/VarselService.kt | 2 +- ...lProducer.kt => ArbeidstakerForhandsvarselProducer.kt} | 8 ++++---- .../no/nav/syfo/application/service/VarselServiceSpek.kt | 4 ++-- 4 files changed, 8 insertions(+), 8 deletions(-) rename src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/{ArbeidstakervarselProducer.kt => ArbeidstakerForhandsvarselProducer.kt} (78%) diff --git a/src/main/kotlin/no/nav/syfo/application/IVarselProducer.kt b/src/main/kotlin/no/nav/syfo/application/IVarselProducer.kt index 70e72238..a1a5117e 100644 --- a/src/main/kotlin/no/nav/syfo/application/IVarselProducer.kt +++ b/src/main/kotlin/no/nav/syfo/application/IVarselProducer.kt @@ -4,5 +4,5 @@ import no.nav.syfo.domain.PersonIdent import no.nav.syfo.domain.Varsel interface IVarselProducer { - fun sendArbeidstakerVarsel(personIdent: PersonIdent, varsel: Varsel) + fun sendArbeidstakerForhandsvarsel(personIdent: PersonIdent, varsel: Varsel) } diff --git a/src/main/kotlin/no/nav/syfo/application/service/VarselService.kt b/src/main/kotlin/no/nav/syfo/application/service/VarselService.kt index 8b3f6784..b5212f7e 100644 --- a/src/main/kotlin/no/nav/syfo/application/service/VarselService.kt +++ b/src/main/kotlin/no/nav/syfo/application/service/VarselService.kt @@ -13,7 +13,7 @@ class VarselService( return unpublishedVarsler.map { (personident, varsel) -> runCatching { - varselProducer.sendArbeidstakerVarsel(personIdent = personident, varsel = varsel) + varselProducer.sendArbeidstakerForhandsvarsel(personIdent = personident, varsel = varsel) val publishedVarsel = varsel.publish() varselRepository.update(publishedVarsel) publishedVarsel diff --git a/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakervarselProducer.kt b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakerForhandsvarselProducer.kt similarity index 78% rename from src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakervarselProducer.kt rename to src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakerForhandsvarselProducer.kt index 3e8eba08..56ffea5a 100644 --- a/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakervarselProducer.kt +++ b/src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/ArbeidstakerForhandsvarselProducer.kt @@ -9,9 +9,9 @@ import org.apache.kafka.clients.producer.KafkaProducer import org.apache.kafka.clients.producer.ProducerRecord import org.slf4j.LoggerFactory -class ArbeidstakervarselProducer(private val kafkaArbeidstakervarselProducer: KafkaProducer) : IVarselProducer { +class ArbeidstakerForhandsvarselProducer(private val kafkaProducer: KafkaProducer) : IVarselProducer { - override fun sendArbeidstakerVarsel(personIdent: PersonIdent, varsel: Varsel) { + override fun sendArbeidstakerForhandsvarsel(personIdent: PersonIdent, varsel: Varsel) { val varselHendelse = ArbeidstakerHendelse( type = HendelseType.SM_ARBEIDSUFORHET_FORHANDSVARSEL, arbeidstakerFnr = personIdent.value, @@ -25,7 +25,7 @@ class ArbeidstakervarselProducer(private val kafkaArbeidstakervarselProducer: Ka ) try { - kafkaArbeidstakervarselProducer.send( + kafkaProducer.send( ProducerRecord( ESYFOVARSEL_TOPIC, UUID.randomUUID().toString(), @@ -40,6 +40,6 @@ class ArbeidstakervarselProducer(private val kafkaArbeidstakervarselProducer: Ka companion object { private const val ESYFOVARSEL_TOPIC = "team-esyfo.varselbus" - private val log = LoggerFactory.getLogger(ArbeidstakervarselProducer::class.java) + private val log = LoggerFactory.getLogger(ArbeidstakerForhandsvarselProducer::class.java) } } diff --git a/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt b/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt index c53d6d87..b885ceff 100644 --- a/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt +++ b/src/test/kotlin/no/nav/syfo/application/service/VarselServiceSpek.kt @@ -8,7 +8,7 @@ import no.nav.syfo.generator.generateForhandsvarselVurdering import no.nav.syfo.infrastructure.database.repository.VarselRepository import no.nav.syfo.infrastructure.database.repository.VurderingRepository import no.nav.syfo.infrastructure.database.dropData -import no.nav.syfo.infrastructure.kafka.esyfovarsel.ArbeidstakervarselProducer +import no.nav.syfo.infrastructure.kafka.esyfovarsel.ArbeidstakerForhandsvarselProducer import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.ArbeidstakerHendelse import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.EsyfovarselHendelse import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.HendelseType @@ -34,7 +34,7 @@ class VarselServiceSpek : Spek({ val vurderingRepository = VurderingRepository(database = database) val kafkaProducer = mockk>() - val varselProducer = ArbeidstakervarselProducer(kafkaArbeidstakervarselProducer = kafkaProducer) + val varselProducer = ArbeidstakerForhandsvarselProducer(kafkaProducer = kafkaProducer) val varselService = VarselService(varselRepository = varselRepository, varselProducer = varselProducer) beforeEachTest {