feat: payload is not a list

maximeperraultdev · maximeperraultdev · commit f440be9ed1eb · 2025-12-16T12:17:22.000+01:00
diff --git a/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/database/model/AISPositionModel.kt b/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/database/model/AISPositionModel.kt
@@ -2,12 +2,8 @@ package fr.gouv.cacem.monitorenv.infrastructure.database.model
 
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 import com.fasterxml.jackson.databind.annotation.JsonSerialize
-import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPositionEntity
-import jakarta.persistence.Entity
-import jakarta.persistence.GeneratedValue
-import jakarta.persistence.GenerationType
-import jakarta.persistence.Id
-import jakarta.persistence.Table
+import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPayload
+import jakarta.persistence.*
 import org.locationtech.jts.geom.Geometry
 import org.locationtech.jts.io.WKTReader
 import org.n52.jackson.datatype.jts.GeometryDeserializer
@@ -31,7 +27,7 @@ data class AISPositionModel(
     val ts: ZonedDateTime?,
 ) {
     companion object {
-        fun toAISPositionModel(aisPosition: AISPositionEntity): AISPositionModel =
+        fun toAISPositionModel(aisPosition: AISPayload): AISPositionModel =
             AISPositionModel(
                 id = aisPosition.id,
                 mmsi = aisPosition.mmsi,
diff --git a/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/database/repositories/JpaAISPositionRepository.kt b/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/database/repositories/JpaAISPositionRepository.kt
@@ -2,7 +2,7 @@ package fr.gouv.cacem.monitorenv.infrastructure.database.repositories
 
 import fr.gouv.cacem.monitorenv.infrastructure.database.model.AISPositionModel
 import fr.gouv.cacem.monitorenv.infrastructure.database.repositories.interfaces.IDBAISPositionRepository
-import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPositionEntity
+import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPayload
 import jakarta.transaction.Transactional
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -15,7 +15,7 @@ class JpaAISPositionRepository(
     private val logger: Logger = LoggerFactory.getLogger(JpaAISPositionRepository::class.java)
 
     @Transactional
-    fun saveAll(aisPositions: List<AISPositionEntity>) {
-        dbAISPositionRepository.saveAll(aisPositions.map { AISPositionModel.toAISPositionModel(it) })
+    fun save(aisPosition: AISPayload) {
+        dbAISPositionRepository.save(AISPositionModel.toAISPositionModel(aisPosition))
     }
 }
diff --git a/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/AISListener.kt b/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/AISListener.kt
@@ -15,16 +15,15 @@ class AISListener(
     private val logger = LoggerFactory.getLogger(AISListener::class.java)
 
     companion object {
-        const val TOPIC = "ais"
+        const val TOPIC = "monitorenv.ais.position"
     }
 
     @KafkaListener(topics = [TOPIC])
     fun listenAIS(payload: AISPayload) {
         try {
-            logger.info("${payload.positions.size} AIS positions received")
-            jpaAISPositionRepository.saveAll(payload.positions)
+            jpaAISPositionRepository.save(payload)
         } catch (ex: Exception) {
-            logger.error(ex.message)
+            logger.error("Could not save ais position:  ${ex.message}")
             throw ex
         }
     }
diff --git a/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/AISProducer.kt b/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/AISProducer.kt
@@ -2,14 +2,13 @@ package fr.gouv.cacem.monitorenv.infrastructure.kafka
 
 import fr.gouv.cacem.monitorenv.infrastructure.kafka.AISListener.Companion.TOPIC
 import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPayload
-import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPositionEntity
 import org.slf4j.LoggerFactory
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Component
 import java.time.ZonedDateTime
-import java.util.*
+import java.util.UUID
 import kotlin.random.Random
 
 /**
@@ -40,19 +39,14 @@ class AISProducer(
             kafkaTemplate.send(
                 TOPIC,
                 AISPayload(
-                    positions =
-                        listOf(
-                            AISPositionEntity(
-                                id = null,
-                                mmsi = Random.nextInt(0, 999999999),
-                                coord = generateRandomPoint(),
-                                status = UUID.randomUUID().toString(),
-                                course = Random.nextDouble(),
-                                heading = Random.nextDouble(),
-                                speed = Random.nextDouble(),
-                                ts = ZonedDateTime.now(),
-                            ),
-                        ),
+                    id = null,
+                    mmsi = Random.nextInt(0, 999999999),
+                    coord = generateRandomPoint(),
+                    status = UUID.randomUUID().toString(),
+                    course = Random.nextDouble(),
+                    heading = Random.nextDouble(),
+                    speed = Random.nextDouble(),
+                    ts = ZonedDateTime.now(),
                 ),
             )
         } catch (ex: Exception) {
diff --git a/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/adapters/AISPayload.kt b/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/adapters/AISPayload.kt
@@ -1,5 +1,14 @@
 package fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters
 
+import java.time.ZonedDateTime
+
 class AISPayload(
-    val positions: List<AISPositionEntity>,
+    val id: Int? = null,
+    val mmsi: Int?,
+    val coord: String?,
+    val status: String?,
+    val course: Double?,
+    val heading: Double?,
+    val speed: Double?,
+    val ts: ZonedDateTime?,
 )
diff --git a/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/adapters/AISPositionEntity.kt b/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/adapters/AISPositionEntity.kt
diff --git a/backend/src/test/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/AISListenerITests.kt b/backend/src/test/kotlin/fr/gouv/cacem/monitorenv/infrastructure/kafka/AISListenerITests.kt
@@ -2,8 +2,8 @@ package fr.gouv.cacem.monitorenv.infrastructure.kafka
 
 import fr.gouv.cacem.monitorenv.infrastructure.database.repositories.AbstractKafkaTests
 import fr.gouv.cacem.monitorenv.infrastructure.database.repositories.interfaces.IDBAISPositionRepository
+import fr.gouv.cacem.monitorenv.infrastructure.kafka.AISListener.Companion.TOPIC
 import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPayload
-import fr.gouv.cacem.monitorenv.infrastructure.kafka.adapters.AISPositionEntity
 import jakarta.transaction.Transactional
 import org.assertj.core.api.Assertions.assertThat
 import org.awaitility.Awaitility
@@ -29,7 +29,7 @@ class AISListenerITests : AbstractKafkaTests() {
         // Given
         val coord = "POINT(-2.7335 47.6078)"
         val aisPosition =
-            AISPositionEntity(
+            AISPayload(
                 id = null,
                 mmsi = 1234567890,
                 coord = coord,
@@ -40,12 +40,12 @@ class AISListenerITests : AbstractKafkaTests() {
                 ts = ZonedDateTime.now(),
             )
 
-        kafkaTemplate.send("ais", AISPayload(positions = listOf(aisPosition))).get(10, TimeUnit.SECONDS)
+        kafkaTemplate.send(TOPIC, aisPosition).get(10, TimeUnit.SECONDS)
 
         Awaitility
             .await()
             .pollInterval(1, TimeUnit.SECONDS)
-            .atMost(30, TimeUnit.SECONDS)
+            .atMost(5, TimeUnit.SECONDS)
             .untilAsserted {
                 val saved = dbAISPositionRepository.findByIdOrNull(1)
                 assertThat(saved).isNotNull()

Original file line number	Diff line number	Diff line change
`@@ -15,16 +15,15 @@ class AISListener(`
`15`	`15`	`private val logger = LoggerFactory.getLogger(AISListener::class.java)`
`16`	`16`
`17`	`17`	`companion object {`
`18`		`- const val TOPIC = "ais"`
	`18`	`+ const val TOPIC = "monitorenv.ais.position"`
`19`	`19`	`}`
`20`	`20`
`21`	`21`	`@KafkaListener(topics = [TOPIC])`
`22`	`22`	`fun listenAIS(payload: AISPayload) {`
`23`	`23`	`try {`
`24`		`- logger.info("${payload.positions.size} AIS positions received")`
`25`		`- jpaAISPositionRepository.saveAll(payload.positions)`
	`24`	`+ jpaAISPositionRepository.save(payload)`
`26`	`25`	`} catch (ex: Exception) {`
`27`		`- logger.error(ex.message)`
	`26`	`+ logger.error("Could not save ais position: ${ex.message}")`
`28`	`27`	`throw ex`
`29`	`28`	`}`
`30`	`29`	`}`