Init
This commit is contained in:
Дмитрий Соловьев
@@ -0,0 +1,35 @@
|
||||
# route-passport-kafka-consumer
|
||||
|
||||
Минимальный сервис на Kotlin + Spring Boot, который:
|
||||
- слушает Kafka topic;
|
||||
- получает в сообщении JSON паспорта маршрута;
|
||||
- парсит JSON в DTO;
|
||||
- больше ничего не делает.
|
||||
|
||||
## Что внутри
|
||||
|
||||
- `RoutePassportConsumerApplication` — точка входа Spring Boot.
|
||||
- `RoutePassportMessageListener` — Kafka listener.
|
||||
- `RoutePassportJsonParserService` — сервис парсинга входного JSON в DTO.
|
||||
- `RoutePassportDto` — итоговый DTO из `pcp-types-lib`.
|
||||
- `RawPassportEnvelopeDto` и связанные `Raw*` DTO — модель входного JSON из `pcp-types-lib`.
|
||||
|
||||
## Как запустить
|
||||
|
||||
1. Поднять Kafka.
|
||||
2. Указать адрес брокера в `application.yml` или через переменную окружения `KAFKA_BOOTSTRAP_SERVERS`.
|
||||
3. При необходимости указать topic через `APP_KAFKA_TOPIC`.
|
||||
4. Запустить приложение:
|
||||
|
||||
```bash
|
||||
./gradlew bootRun
|
||||
```
|
||||
|
||||
## Сообщение, которое ожидает сервис
|
||||
|
||||
Сервис ожидает, что value Kafka-сообщения — это JSON того же формата, что и `src/test/resources/passport_example.json`.
|
||||
Поле `geometry` должно приходить строкой в формате WKT.
|
||||
|
||||
## Примечание
|
||||
|
||||
Поскольку в запросе был указан просто «слушает очередь», реализация сделана через Kafka topic как наиболее типичный вариант для стека Kotlin + Spring и Kafka-ориентированного сервиса.
|
||||
@@ -0,0 +1,57 @@
|
||||
group = "space.nstart.pcp"
|
||||
|
||||
plugins {
|
||||
kotlin("jvm")
|
||||
kotlin("plugin.spring")
|
||||
kotlin("plugin.jpa")
|
||||
id("org.springframework.boot")
|
||||
id("io.spring.dependency-management")
|
||||
}
|
||||
|
||||
version = "1.0.0"
|
||||
|
||||
description = "Входная точка маршрутов от портала для их последующей обработки в ПЦП"
|
||||
|
||||
kotlin {
|
||||
jvmToolchain((property("versions.java") as String).toInt())
|
||||
compilerOptions {
|
||||
freeCompilerArgs.addAll("-Xjsr305=strict")
|
||||
jvmTarget.set(org.jetbrains.kotlin.gradle.dsl.JvmTarget.fromTarget(property("versions.java") as String))
|
||||
}
|
||||
}
|
||||
|
||||
dependencies {
|
||||
implementation(project(":libs:pcp-types-lib"))
|
||||
|
||||
implementation("${property("dep.spring.actuator")}")
|
||||
implementation("org.springframework.boot:spring-boot-starter")
|
||||
implementation("org.springframework.boot:spring-boot-starter-web")
|
||||
implementation("org.springframework.boot:spring-boot-starter-data-jpa")
|
||||
implementation("org.springframework.boot:spring-boot-starter-flyway")
|
||||
implementation("org.springframework.boot:spring-boot-starter-validation")
|
||||
implementation("org.flywaydb:flyway-database-postgresql")
|
||||
implementation("org.springdoc:springdoc-openapi-starter-webmvc-ui:${property("versions.open-api")}")
|
||||
implementation("org.springframework.kafka:spring-kafka")
|
||||
implementation("org.springframework.cloud:spring-cloud-starter-config")
|
||||
implementation("tools.jackson.module:jackson-module-kotlin")
|
||||
implementation("org.locationtech.jts:jts-core:1.19.0")
|
||||
implementation("org.jetbrains.kotlin:kotlin-reflect")
|
||||
|
||||
runtimeOnly("org.postgresql:postgresql")
|
||||
|
||||
testImplementation("org.jetbrains.kotlin:kotlin-test-junit5")
|
||||
testImplementation("org.springframework.boot:spring-boot-starter-test")
|
||||
testImplementation("org.springframework.kafka:spring-kafka-test")
|
||||
|
||||
testRuntimeOnly("org.junit.platform:junit-platform-launcher")
|
||||
}
|
||||
|
||||
dependencyManagement {
|
||||
imports {
|
||||
mavenBom("org.springframework.cloud:spring-cloud-dependencies:${property("versions.spring.cloud")}")
|
||||
}
|
||||
}
|
||||
|
||||
tasks.withType<Test> {
|
||||
useJUnitPlatform()
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
+25
@@ -0,0 +1,25 @@
|
||||
package space.nstart.pcp.routepassportconsumer
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication
|
||||
import org.springframework.boot.context.properties.ConfigurationPropertiesScan
|
||||
import org.springframework.boot.runApplication
|
||||
import org.springframework.kafka.annotation.EnableKafka
|
||||
|
||||
/**
|
||||
* Главный класс Spring Boot приложения.
|
||||
*
|
||||
* Приложение поднимает только инфраструктуру потребителя Kafka и сервис парсинга JSON.
|
||||
*/
|
||||
@ConfigurationPropertiesScan
|
||||
@SpringBootApplication
|
||||
@EnableKafka
|
||||
class RoutePassportConsumerApplication
|
||||
|
||||
/**
|
||||
* Точка входа приложения.
|
||||
*
|
||||
* @param args аргументы командной строки, которые передаются в Spring Boot.
|
||||
*/
|
||||
fun main(args: Array<String>) {
|
||||
runApplication<RoutePassportConsumerApplication>(*args)
|
||||
}
|
||||
+131
@@ -0,0 +1,131 @@
|
||||
package space.nstart.pcp.routepassportconsumer.config
|
||||
|
||||
import com.fasterxml.jackson.core.JsonParseException
|
||||
import com.fasterxml.jackson.core.JsonProcessingException
|
||||
import com.fasterxml.jackson.databind.JsonMappingException
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord
|
||||
import org.apache.kafka.common.TopicPartition
|
||||
import org.apache.kafka.common.header.Headers
|
||||
import org.apache.kafka.common.header.internals.RecordHeader
|
||||
import org.apache.kafka.common.header.internals.RecordHeaders
|
||||
import org.apache.kafka.common.serialization.StringDeserializer
|
||||
import org.springframework.beans.factory.annotation.Value
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
|
||||
import org.springframework.context.annotation.Bean
|
||||
import org.springframework.context.annotation.Configuration
|
||||
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
|
||||
import org.springframework.kafka.core.ConsumerFactory
|
||||
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
|
||||
import org.springframework.kafka.core.KafkaTemplate
|
||||
import org.springframework.kafka.listener.ContainerProperties
|
||||
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer
|
||||
import org.springframework.kafka.listener.DefaultErrorHandler
|
||||
import org.springframework.util.backoff.ExponentialBackOff
|
||||
import space.nstart.pcp.routepassportconsumer.exception.RoutePassportContractException
|
||||
import tools.jackson.core.JacksonException
|
||||
import java.nio.charset.StandardCharsets
|
||||
|
||||
@Configuration
|
||||
@ConditionalOnProperty(name = ["spring.kafka.bootstrap-servers"], matchIfMissing = false)
|
||||
class KafkaConsumerConfig(
|
||||
@param:Value("\${spring.kafka.bootstrap-servers}")
|
||||
private val bootstrapServers: String,
|
||||
@param:Value("\${spring.kafka.consumer.group-id}")
|
||||
private val groupId: String,
|
||||
@param:Value("\${spring.kafka.consumer.auto-offset-reset:latest}")
|
||||
private val autoOffsetReset: String,
|
||||
) {
|
||||
|
||||
@Bean
|
||||
fun consumerFactory(): ConsumerFactory<String, String> =
|
||||
DefaultKafkaConsumerFactory(
|
||||
mutableMapOf<String, Any>(
|
||||
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
|
||||
ConsumerConfig.GROUP_ID_CONFIG to groupId,
|
||||
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to autoOffsetReset,
|
||||
ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false,
|
||||
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
|
||||
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
|
||||
),
|
||||
)
|
||||
|
||||
@Bean
|
||||
fun kafkaListenerContainerFactory(
|
||||
kafkaErrorHandler: DefaultErrorHandler,
|
||||
): ConcurrentKafkaListenerContainerFactory<String, String> =
|
||||
ConcurrentKafkaListenerContainerFactory<String, String>().apply {
|
||||
setConsumerFactory(consumerFactory())
|
||||
setCommonErrorHandler(kafkaErrorHandler)
|
||||
containerProperties.ackMode = ContainerProperties.AckMode.MANUAL_IMMEDIATE
|
||||
}
|
||||
|
||||
@Bean
|
||||
fun deadLetterPublishingRecoverer(
|
||||
kafkaTemplate: KafkaTemplate<String, String>,
|
||||
kafkaTopicsProperties: KafkaTopicsProperties,
|
||||
): DeadLetterPublishingRecoverer =
|
||||
DeadLetterPublishingRecoverer(kafkaTemplate) { record, _ ->
|
||||
TopicPartition(kafkaTopicsProperties.inputDlq, record.partition())
|
||||
}.apply {
|
||||
addHeadersFunction { record, exception -> routePassportDlqHeaders(record, exception) }
|
||||
}
|
||||
|
||||
@Bean
|
||||
fun kafkaErrorHandler(
|
||||
deadLetterPublishingRecoverer: DeadLetterPublishingRecoverer,
|
||||
retryProperties: RoutePassportConsumerRetryProperties,
|
||||
): DefaultErrorHandler =
|
||||
DefaultErrorHandler(
|
||||
deadLetterPublishingRecoverer,
|
||||
routePassportConsumerBackOff(retryProperties),
|
||||
).apply {
|
||||
addNotRetryableExceptions(
|
||||
RoutePassportContractException::class.java,
|
||||
JacksonException::class.java,
|
||||
JsonProcessingException::class.java,
|
||||
JsonMappingException::class.java,
|
||||
JsonParseException::class.java,
|
||||
)
|
||||
setCommitRecovered(true)
|
||||
}
|
||||
|
||||
private fun routePassportConsumerBackOff(
|
||||
retryProperties: RoutePassportConsumerRetryProperties,
|
||||
): ExponentialBackOff =
|
||||
ExponentialBackOff().apply {
|
||||
initialInterval = retryProperties.backoffMs
|
||||
multiplier = RETRY_BACKOFF_MULTIPLIER
|
||||
maxInterval = retryProperties.maxBackoffMs
|
||||
maxAttempts = (retryProperties.maxAttempts - 1).coerceAtLeast(0)
|
||||
}
|
||||
|
||||
private fun routePassportDlqHeaders(
|
||||
record: ConsumerRecord<*, *>,
|
||||
exception: Exception,
|
||||
): Headers =
|
||||
RecordHeaders().apply {
|
||||
addStringHeader(HEADER_SERVICE_NAME, SERVICE_NAME)
|
||||
addStringHeader(HEADER_EXCEPTION_CLASS, exception::class.java.name)
|
||||
addStringHeader(HEADER_EXCEPTION_MESSAGE, exception.message.orEmpty())
|
||||
addStringHeader(HEADER_ORIGINAL_TOPIC, record.topic())
|
||||
addStringHeader(HEADER_ORIGINAL_PARTITION, record.partition().toString())
|
||||
addStringHeader(HEADER_ORIGINAL_OFFSET, record.offset().toString())
|
||||
}
|
||||
|
||||
private fun Headers.addStringHeader(name: String, value: String) {
|
||||
add(RecordHeader(name, value.toByteArray(StandardCharsets.UTF_8)))
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val SERVICE_NAME = "pcp-route-processing-service"
|
||||
const val RETRY_BACKOFF_MULTIPLIER = 2.0
|
||||
|
||||
const val HEADER_SERVICE_NAME = "pcp-service-name"
|
||||
const val HEADER_EXCEPTION_CLASS = "pcp-exception-class"
|
||||
const val HEADER_EXCEPTION_MESSAGE = "pcp-exception-message"
|
||||
const val HEADER_ORIGINAL_TOPIC = "pcp-original-topic"
|
||||
const val HEADER_ORIGINAL_PARTITION = "pcp-original-partition"
|
||||
const val HEADER_ORIGINAL_OFFSET = "pcp-original-offset"
|
||||
}
|
||||
}
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
package space.nstart.pcp.routepassportconsumer.config
|
||||
|
||||
import org.apache.kafka.clients.producer.ProducerConfig
|
||||
import org.apache.kafka.common.serialization.StringSerializer
|
||||
import org.springframework.beans.factory.annotation.Value
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
|
||||
import org.springframework.context.annotation.Bean
|
||||
import org.springframework.context.annotation.Configuration
|
||||
import org.springframework.kafka.core.DefaultKafkaProducerFactory
|
||||
import org.springframework.kafka.core.KafkaTemplate
|
||||
import org.springframework.kafka.core.ProducerFactory
|
||||
|
||||
/**
|
||||
* Явная producer-конфигурация для route-processing-service.
|
||||
*
|
||||
* В этом модуле Kafka producer bean'ы не поднимались автоконфигурацией,
|
||||
* поэтому template регистрируется явно.
|
||||
*/
|
||||
@Configuration
|
||||
@ConditionalOnProperty(name = ["spring.kafka.bootstrap-servers"], matchIfMissing = false)
|
||||
class KafkaProducerConfig(
|
||||
@param:Value("\${spring.kafka.bootstrap-servers}")
|
||||
private val bootstrapServers: String,
|
||||
) {
|
||||
|
||||
@Bean
|
||||
fun producerFactory(): ProducerFactory<String, String> =
|
||||
DefaultKafkaProducerFactory(
|
||||
mutableMapOf<String, Any>(
|
||||
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
|
||||
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
|
||||
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
|
||||
),
|
||||
)
|
||||
|
||||
@Bean
|
||||
fun kafkaTemplate(): KafkaTemplate<String, String> = KafkaTemplate(producerFactory())
|
||||
}
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
package space.nstart.pcp.routepassportconsumer.config
|
||||
|
||||
import jakarta.validation.constraints.NotBlank
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties
|
||||
import org.springframework.validation.annotation.Validated
|
||||
|
||||
/**
|
||||
* Типизированные Kafka topic-настройки сервиса.
|
||||
*
|
||||
* Все значения приходят из spring-cloud-config-server и валидируются при старте,
|
||||
* чтобы сервис не запускался без обязательных имен входного и выходных topic'ов.
|
||||
*/
|
||||
@Validated
|
||||
@ConfigurationProperties(prefix = "app.kafka.topics")
|
||||
data class KafkaTopicsProperties(
|
||||
|
||||
/** Topic, из которого сервис читает входящие паспорта маршрутов. */
|
||||
@field:NotBlank
|
||||
val input: String = "",
|
||||
|
||||
/** DLQ topic для входящих паспортов маршрутов, которые не удалось обработать. */
|
||||
@field:NotBlank
|
||||
val inputDlq: String = "",
|
||||
|
||||
/** Topic, в который сервис будет публиковать сообщение об обработанном маршруте. */
|
||||
@field:NotBlank
|
||||
val processedRoute: String = "",
|
||||
|
||||
/** Topic, в который сервис будет публиковать сообщение геопривязки маршрута. */
|
||||
@field:NotBlank
|
||||
val routeGeoreference: String = "",
|
||||
)
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
package space.nstart.pcp.routepassportconsumer.config
|
||||
|
||||
import io.swagger.v3.oas.models.OpenAPI
|
||||
import io.swagger.v3.oas.models.info.Info
|
||||
import org.springframework.context.annotation.Bean
|
||||
import org.springframework.context.annotation.Configuration
|
||||
|
||||
/**
|
||||
* Базовая OpenAPI-конфигурация для route-processing-service.
|
||||
*/
|
||||
@Configuration
|
||||
class OpenApiConfig {
|
||||
@Bean
|
||||
fun routeProcessingOpenApi(): OpenAPI {
|
||||
return OpenAPI()
|
||||
.info(
|
||||
Info()
|
||||
.title("PCP Route Processing Service API")
|
||||
.description("REST API сервиса обработки паспортов маршрутов PCP.")
|
||||
.version("v1"),
|
||||
)
|
||||
}
|
||||
}
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
package space.nstart.pcp.routepassportconsumer.config
|
||||
|
||||
import jakarta.validation.constraints.Min
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties
|
||||
import org.springframework.validation.annotation.Validated
|
||||
|
||||
/**
|
||||
* Настройки bounded retry для input Kafka consumer route passport flow.
|
||||
*/
|
||||
@Validated
|
||||
@ConfigurationProperties(prefix = "app.route-passport.consumer.retry")
|
||||
data class RoutePassportConsumerRetryProperties(
|
||||
|
||||
/** Максимальное количество попыток обработки, включая первичную delivery. */
|
||||
@field:Min(1)
|
||||
val maxAttempts: Long = 3,
|
||||
|
||||
/** Начальная задержка retry в миллисекундах. */
|
||||
@field:Min(0)
|
||||
val backoffMs: Long = 1_000,
|
||||
|
||||
/** Максимальная задержка между retry в миллисекундах. */
|
||||
@field:Min(0)
|
||||
val maxBackoffMs: Long = 30_000,
|
||||
)
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
package space.nstart.pcp.routepassportconsumer.controller
|
||||
|
||||
import io.swagger.v3.oas.annotations.Operation
|
||||
import io.swagger.v3.oas.annotations.tags.Tag
|
||||
import org.springframework.core.io.ClassPathResource
|
||||
import org.springframework.kafka.core.KafkaTemplate
|
||||
import org.springframework.web.bind.annotation.PostMapping
|
||||
import org.springframework.web.bind.annotation.GetMapping
|
||||
import org.springframework.web.bind.annotation.RequestMapping
|
||||
import org.springframework.web.bind.annotation.RestController
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import java.nio.charset.StandardCharsets
|
||||
|
||||
@RestController
|
||||
@RequestMapping("/api/route-processing")
|
||||
@Tag(name = "Route Processing", description = "Тестовые endpoint'ы сервиса обработки маршрутов")
|
||||
class RouteProcessingTestController(
|
||||
private val kafkaTemplate: KafkaTemplate<String, String>,
|
||||
private val kafkaTopicsProperties: KafkaTopicsProperties,
|
||||
) {
|
||||
|
||||
@GetMapping("/test")
|
||||
@Operation(summary = "Проверка доступности сервиса")
|
||||
fun test(): Map<String, String> = mapOf(
|
||||
"status" to "ok",
|
||||
"service" to "pcp-route-processing-service",
|
||||
)
|
||||
|
||||
@PostMapping("/test/send-passport-message")
|
||||
@Operation(summary = "Отправить тестовый паспорт маршрута в Kafka")
|
||||
fun sendPassportMessage(): Map<String, String> {
|
||||
val payload = ClassPathResource("test-passport-message.json")
|
||||
.inputStream
|
||||
.readAllBytes()
|
||||
.toString(StandardCharsets.UTF_8)
|
||||
|
||||
kafkaTemplate.send(kafkaTopicsProperties.input, payload)
|
||||
|
||||
return mapOf(
|
||||
"status" to "sent",
|
||||
"topic" to kafkaTopicsProperties.input,
|
||||
"resource" to "classpath:test-passport-message.json",
|
||||
)
|
||||
}
|
||||
}
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
package space.nstart.pcp.routepassportconsumer.entity
|
||||
|
||||
import jakarta.persistence.Column
|
||||
import jakarta.persistence.Entity
|
||||
import jakarta.persistence.Id
|
||||
import jakarta.persistence.Table
|
||||
import jakarta.persistence.Version
|
||||
import org.hibernate.annotations.JdbcTypeCode
|
||||
import org.hibernate.type.SqlTypes
|
||||
import tools.jackson.databind.JsonNode
|
||||
import java.math.BigDecimal
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
@Entity
|
||||
@Table(name = "route_passports")
|
||||
class RoutePassportEntity(
|
||||
@Id
|
||||
@Column(name = "route_id", nullable = false)
|
||||
var routeId: UUID,
|
||||
|
||||
@Column(name = "route_name_full", nullable = false, columnDefinition = "TEXT")
|
||||
var routeNameFull: String,
|
||||
|
||||
@Column(name = "route_name_short", nullable = false, length = 255)
|
||||
var routeNameShort: String,
|
||||
|
||||
@Column(name = "ka_short", nullable = false, length = 64)
|
||||
var kaShort: String,
|
||||
|
||||
@Column(name = "ka_full", nullable = false, length = 255)
|
||||
var kaFull: String,
|
||||
|
||||
@Column(name = "orbit_number", nullable = false)
|
||||
var orbitNumber: Long,
|
||||
|
||||
@Column(name = "orbit_state", nullable = false, length = 64)
|
||||
var orbitState: String,
|
||||
|
||||
@Column(name = "interval_begin", nullable = false)
|
||||
var intervalBegin: LocalDateTime,
|
||||
|
||||
@Column(name = "interval_end", nullable = false)
|
||||
var intervalEnd: LocalDateTime,
|
||||
|
||||
@Column(name = "roll_angle_min", nullable = false)
|
||||
var rollAngleMin: BigDecimal,
|
||||
|
||||
@Column(name = "roll_angle_max", nullable = false)
|
||||
var rollAngleMax: BigDecimal,
|
||||
|
||||
@Column(name = "visir_angle_min", nullable = false)
|
||||
var visirAngleMin: BigDecimal,
|
||||
|
||||
@Column(name = "visir_angle_max", nullable = false)
|
||||
var visirAngleMax: BigDecimal,
|
||||
|
||||
@Column(name = "resolution_range", nullable = false)
|
||||
var resolutionRange: BigDecimal,
|
||||
|
||||
@Column(name = "resolution_azimuth", nullable = false)
|
||||
var resolutionAzimuth: BigDecimal,
|
||||
|
||||
@JdbcTypeCode(SqlTypes.JSON)
|
||||
@Column(name = "polarisation", nullable = false, columnDefinition = "jsonb")
|
||||
var polarisation: JsonNode,
|
||||
|
||||
@Column(name = "processing_level", nullable = false, length = 64)
|
||||
var processingLevel: String,
|
||||
|
||||
@Column(name = "geometry_wkt", nullable = false, columnDefinition = "TEXT")
|
||||
var geometryWkt: String,
|
||||
|
||||
@Column(name = "route_status", nullable = false, length = 32)
|
||||
var routeStatus: String,
|
||||
|
||||
@Column(name = "trace_id", length = 255)
|
||||
var traceId: String? = null,
|
||||
|
||||
@Column(name = "correlation_id", length = 255)
|
||||
var correlationId: String? = null,
|
||||
|
||||
@Column(name = "source_topic", length = 255)
|
||||
var sourceTopic: String? = null,
|
||||
|
||||
@Column(name = "source_partition")
|
||||
var sourcePartition: Int? = null,
|
||||
|
||||
@Column(name = "source_offset")
|
||||
var sourceOffset: Long? = null,
|
||||
|
||||
@Column(name = "source_key", length = 255)
|
||||
var sourceKey: String? = null,
|
||||
|
||||
@JdbcTypeCode(SqlTypes.JSON)
|
||||
@Column(name = "raw_payload", columnDefinition = "jsonb")
|
||||
var rawPayload: JsonNode? = null,
|
||||
|
||||
@Column(name = "created_at", nullable = false)
|
||||
var createdAt: LocalDateTime,
|
||||
|
||||
@Column(name = "updated_at", nullable = false)
|
||||
var updatedAt: LocalDateTime,
|
||||
|
||||
@Version
|
||||
@Column(name = "version", nullable = false)
|
||||
var version: Long = 0,
|
||||
)
|
||||
+69
@@ -0,0 +1,69 @@
|
||||
package space.nstart.pcp.routepassportconsumer.entity
|
||||
|
||||
import jakarta.persistence.Column
|
||||
import jakarta.persistence.Entity
|
||||
import jakarta.persistence.EnumType
|
||||
import jakarta.persistence.Enumerated
|
||||
import jakarta.persistence.Id
|
||||
import jakarta.persistence.Table
|
||||
import org.hibernate.annotations.JdbcTypeCode
|
||||
import org.hibernate.type.SqlTypes
|
||||
import tools.jackson.databind.JsonNode
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
@Entity
|
||||
@Table(name = "route_passport_outbox")
|
||||
class RoutePassportOutboxEntity(
|
||||
@Id
|
||||
@Column(name = "id", nullable = false)
|
||||
var id: UUID,
|
||||
|
||||
@Column(name = "route_id", nullable = false)
|
||||
var routeId: UUID,
|
||||
|
||||
@Column(name = "event_type", nullable = false, length = 128)
|
||||
var eventType: String,
|
||||
|
||||
@Column(name = "topic", nullable = false, length = 255)
|
||||
var topic: String,
|
||||
|
||||
@Column(name = "message_key", nullable = false, length = 255)
|
||||
var messageKey: String,
|
||||
|
||||
@JdbcTypeCode(SqlTypes.JSON)
|
||||
@Column(name = "payload", nullable = false, columnDefinition = "jsonb")
|
||||
var payload: JsonNode,
|
||||
|
||||
@JdbcTypeCode(SqlTypes.JSON)
|
||||
@Column(name = "headers", columnDefinition = "jsonb")
|
||||
var headers: JsonNode? = null,
|
||||
|
||||
@Enumerated(EnumType.STRING)
|
||||
@Column(name = "status", nullable = false, length = 32)
|
||||
var status: RoutePassportOutboxStatus,
|
||||
|
||||
@Column(name = "attempts", nullable = false)
|
||||
var attempts: Int,
|
||||
|
||||
@Column(name = "max_attempts", nullable = false)
|
||||
var maxAttempts: Int,
|
||||
|
||||
@Column(name = "next_attempt_at", nullable = false)
|
||||
var nextAttemptAt: LocalDateTime,
|
||||
|
||||
@Column(name = "locked_until")
|
||||
var lockedUntil: LocalDateTime? = null,
|
||||
|
||||
@Column(name = "last_error", columnDefinition = "TEXT")
|
||||
var lastError: String? = null,
|
||||
|
||||
@Column(name = "created_at", nullable = false)
|
||||
var createdAt: LocalDateTime,
|
||||
|
||||
@Column(name = "updated_at", nullable = false)
|
||||
var updatedAt: LocalDateTime,
|
||||
|
||||
@Column(name = "published_at")
|
||||
var publishedAt: LocalDateTime? = null,
|
||||
)
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
package space.nstart.pcp.routepassportconsumer.entity
|
||||
|
||||
/**
|
||||
* Состояние события outbox в жизненном цикле публикации.
|
||||
*/
|
||||
enum class RoutePassportOutboxStatus {
|
||||
PENDING,
|
||||
PUBLISHED,
|
||||
FAILED,
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
package space.nstart.pcp.routepassportconsumer.exception
|
||||
|
||||
class RoutePassportContractException(
|
||||
message: String,
|
||||
cause: Throwable? = null,
|
||||
) : RuntimeException(message, cause)
|
||||
+95
@@ -0,0 +1,95 @@
|
||||
package space.nstart.pcp.routepassportconsumer.listener
|
||||
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord
|
||||
import org.slf4j.LoggerFactory
|
||||
import org.springframework.kafka.annotation.KafkaListener
|
||||
import org.springframework.kafka.support.Acknowledgment
|
||||
import org.springframework.stereotype.Component
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import space.nstart.pcp.routepassportconsumer.service.RoutePassportIngestionService
|
||||
import space.nstart.pcp.routepassportconsumer.service.RoutePassportJsonParserService
|
||||
import space.nstart.pcp.routepassportconsumer.service.RoutePassportSourceRecordMetadata
|
||||
|
||||
/**
|
||||
* Kafka listener, который получает JSON паспорта маршрута и передаёт его в транзакционный ingestion.
|
||||
*/
|
||||
@Component
|
||||
class RoutePassportMessageListener(
|
||||
|
||||
/** Сервис парсинга входного JSON сообщения в итоговый DTO. */
|
||||
private val routePassportJsonParserService: RoutePassportJsonParserService,
|
||||
|
||||
/** Сервис транзакционного сохранения паспорта маршрута и outbox-событий. */
|
||||
private val routePassportIngestionService: RoutePassportIngestionService,
|
||||
|
||||
/** Набор Kafka topic'ов, который приходит из config server и валидируется при старте. */
|
||||
val kafkaTopicsProperties: KafkaTopicsProperties,
|
||||
) {
|
||||
private val log = LoggerFactory.getLogger(this::class.java)
|
||||
|
||||
/**
|
||||
* Обрабатывает входное сообщение Kafka.
|
||||
*
|
||||
* Offset подтверждается только после успешного возврата из transactional ingestion.
|
||||
*/
|
||||
@KafkaListener(
|
||||
topics = ["#{__listener.kafkaTopicsProperties.input}"],
|
||||
groupId = "\${spring.kafka.consumer.group-id}",
|
||||
)
|
||||
fun listen(
|
||||
record: ConsumerRecord<String, String>,
|
||||
acknowledgment: Acknowledgment,
|
||||
) {
|
||||
val rawPayload = record.value()
|
||||
val sourceRecord = RoutePassportSourceRecordMetadata(
|
||||
topic = record.topic(),
|
||||
partition = record.partition(),
|
||||
offset = record.offset(),
|
||||
key = record.key(),
|
||||
)
|
||||
|
||||
try {
|
||||
log.debug(
|
||||
"Kafka record received: sourceTopic={}, partition={}, offset={}, key={}, payloadSize={}",
|
||||
sourceRecord.topic,
|
||||
sourceRecord.partition,
|
||||
sourceRecord.offset,
|
||||
sourceRecord.key,
|
||||
rawPayload.length,
|
||||
)
|
||||
|
||||
val parsedMessage = routePassportJsonParserService.parse(rawPayload)
|
||||
routePassportIngestionService.ingest(
|
||||
parsedMessage = parsedMessage,
|
||||
rawPayload = rawPayload,
|
||||
sourceRecord = sourceRecord,
|
||||
)
|
||||
|
||||
// Transactional ingest commits before returning; only then the Kafka offset is acknowledged.
|
||||
acknowledgment.acknowledge()
|
||||
|
||||
log.info(
|
||||
"Route passport ingested: sourceTopic={}, partition={}, offset={}, key={}, payloadSize={}, routeId={}, traceId={}, correlationId={}",
|
||||
sourceRecord.topic,
|
||||
sourceRecord.partition,
|
||||
sourceRecord.offset,
|
||||
sourceRecord.key,
|
||||
rawPayload.length,
|
||||
parsedMessage.routePassportDto.routeId,
|
||||
parsedMessage.metadata.traceId,
|
||||
parsedMessage.metadata.correlationId,
|
||||
)
|
||||
} catch (exception: Exception) {
|
||||
log.error(
|
||||
"Failed to process Kafka record: sourceTopic={}, partition={}, offset={}, key={}, error={}",
|
||||
sourceRecord.topic,
|
||||
sourceRecord.partition,
|
||||
sourceRecord.offset,
|
||||
sourceRecord.key,
|
||||
exception.message,
|
||||
exception,
|
||||
)
|
||||
throw exception
|
||||
}
|
||||
}
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
package space.nstart.pcp.routepassportconsumer.repository
|
||||
|
||||
import org.springframework.data.jpa.repository.JpaRepository
|
||||
import org.springframework.data.jpa.repository.Query
|
||||
import org.springframework.data.repository.query.Param
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
|
||||
import java.util.UUID
|
||||
|
||||
interface RoutePassportOutboxRepository : JpaRepository<RoutePassportOutboxEntity, UUID> {
|
||||
fun existsByRouteIdAndEventType(routeId: UUID, eventType: String): Boolean
|
||||
|
||||
@Query(
|
||||
value = """
|
||||
SELECT *
|
||||
FROM route_passport_outbox
|
||||
WHERE status = 'PENDING'
|
||||
AND next_attempt_at <= now()
|
||||
AND (locked_until IS NULL OR locked_until < now())
|
||||
ORDER BY created_at
|
||||
LIMIT :limit
|
||||
FOR UPDATE SKIP LOCKED
|
||||
""",
|
||||
nativeQuery = true,
|
||||
)
|
||||
fun findDueForUpdateSkipLocked(@Param("limit") limit: Int): List<RoutePassportOutboxEntity>
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
package space.nstart.pcp.routepassportconsumer.repository
|
||||
|
||||
import org.springframework.data.jpa.repository.JpaRepository
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportEntity
|
||||
import java.util.UUID
|
||||
|
||||
interface RoutePassportRepository : JpaRepository<RoutePassportEntity, UUID>
|
||||
+143
@@ -0,0 +1,143 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.locationtech.jts.geom.Coordinate
|
||||
import org.locationtech.jts.geom.Geometry
|
||||
import org.locationtech.jts.geom.GeometryFactory
|
||||
import org.locationtech.jts.geom.LinearRing
|
||||
import org.locationtech.jts.geom.Polygon
|
||||
import org.locationtech.jts.io.WKTWriter
|
||||
import org.springframework.stereotype.Service
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RawGeoJsonGeometryDto
|
||||
import space.nstart.pcp.routepassportconsumer.exception.RoutePassportContractException
|
||||
import tools.jackson.databind.JsonNode
|
||||
|
||||
/**
|
||||
* Конвертирует поддерживаемую GeoJSON geometry входного паспорта маршрута в WKT.
|
||||
*/
|
||||
@Service
|
||||
open class RouteGeometryConverter {
|
||||
private val geometryFactory = GeometryFactory()
|
||||
private val wktWriter = WKTWriter()
|
||||
|
||||
open fun convert(geometry: RawGeoJsonGeometryDto?): String {
|
||||
if (geometry == null) {
|
||||
throw RoutePassportContractException("Отсутствует обязательное поле geometry")
|
||||
}
|
||||
|
||||
val jtsGeometry = when (geometry.type) {
|
||||
"Polygon" -> toPolygon(geometry.coordinates, fieldName = "geometry.coordinates")
|
||||
"MultiPolygon" -> toMultiPolygon(geometry.coordinates)
|
||||
else -> throw RoutePassportContractException("Неподдерживаемый тип geometry: ${geometry.type}")
|
||||
}
|
||||
|
||||
return writeCompactWkt(jtsGeometry)
|
||||
}
|
||||
|
||||
private fun toMultiPolygon(coordinatesNode: JsonNode): Geometry {
|
||||
requireArray(coordinatesNode, "geometry.coordinates")
|
||||
|
||||
if (coordinatesNode.size() == 0) {
|
||||
throw RoutePassportContractException("MultiPolygon geometry.coordinates не должен быть пустым")
|
||||
}
|
||||
|
||||
val polygons = coordinatesNode.mapIndexed { index, polygonNode ->
|
||||
toPolygon(polygonNode, fieldName = "geometry.coordinates[$index]")
|
||||
}.toTypedArray()
|
||||
|
||||
return geometryFactory.createMultiPolygon(polygons)
|
||||
}
|
||||
|
||||
private fun toPolygon(
|
||||
coordinatesNode: JsonNode,
|
||||
fieldName: String,
|
||||
): Polygon {
|
||||
requireArray(coordinatesNode, fieldName)
|
||||
|
||||
if (coordinatesNode.size() == 0) {
|
||||
throw RoutePassportContractException("Polygon $fieldName должен содержать хотя бы один linear ring")
|
||||
}
|
||||
|
||||
val shell = toLinearRing(coordinatesNode[0], "$fieldName[0]")
|
||||
val holes = coordinatesNode.drop(1).mapIndexed { index, ringNode ->
|
||||
toLinearRing(ringNode, "$fieldName[${index + 1}]")
|
||||
}.toTypedArray()
|
||||
|
||||
return try {
|
||||
geometryFactory.createPolygon(shell, holes)
|
||||
} catch (exception: IllegalArgumentException) {
|
||||
throw RoutePassportContractException("Некорректная структура Polygon $fieldName", exception)
|
||||
}
|
||||
}
|
||||
|
||||
private fun toLinearRing(
|
||||
ringNode: JsonNode,
|
||||
fieldName: String,
|
||||
): LinearRing {
|
||||
requireArray(ringNode, fieldName)
|
||||
|
||||
if (ringNode.size() < MIN_LINEAR_RING_POINTS) {
|
||||
throw RoutePassportContractException(
|
||||
"Linear ring $fieldName должен содержать минимум $MIN_LINEAR_RING_POINTS точки",
|
||||
)
|
||||
}
|
||||
|
||||
val coordinates = ringNode.mapIndexed { index, coordinateNode ->
|
||||
toCoordinate(coordinateNode, "$fieldName[$index]")
|
||||
}.toTypedArray()
|
||||
|
||||
if (!coordinates.first().equals2D(coordinates.last())) {
|
||||
throw RoutePassportContractException("Linear ring $fieldName должен быть замкнут")
|
||||
}
|
||||
|
||||
return try {
|
||||
geometryFactory.createLinearRing(coordinates)
|
||||
} catch (exception: IllegalArgumentException) {
|
||||
throw RoutePassportContractException("Некорректная структура linear ring $fieldName", exception)
|
||||
}
|
||||
}
|
||||
|
||||
private fun toCoordinate(
|
||||
coordinateNode: JsonNode,
|
||||
fieldName: String,
|
||||
): Coordinate {
|
||||
requireArray(coordinateNode, fieldName)
|
||||
|
||||
if (coordinateNode.size() < MIN_COORDINATE_VALUES) {
|
||||
throw RoutePassportContractException("Точка $fieldName должна содержать минимум две координаты")
|
||||
}
|
||||
|
||||
if (!coordinateNode[0].isNumber || !coordinateNode[1].isNumber) {
|
||||
throw RoutePassportContractException("Точка $fieldName должна содержать числовые longitude и latitude")
|
||||
}
|
||||
|
||||
return try {
|
||||
Coordinate(
|
||||
coordinateNode[0].decimalValue().toDouble(),
|
||||
coordinateNode[1].decimalValue().toDouble(),
|
||||
)
|
||||
} catch (exception: RuntimeException) {
|
||||
throw RoutePassportContractException("Точка $fieldName содержит некорректные координаты", exception)
|
||||
}
|
||||
}
|
||||
|
||||
private fun requireArray(
|
||||
node: JsonNode,
|
||||
fieldName: String,
|
||||
) {
|
||||
if (!node.isArray) {
|
||||
throw RoutePassportContractException("Поле $fieldName должно быть массивом GeoJSON")
|
||||
}
|
||||
}
|
||||
|
||||
private fun writeCompactWkt(geometry: Geometry): String =
|
||||
when (geometry.geometryType) {
|
||||
"Polygon" -> wktWriter.write(geometry).removePrefix("POLYGON ").let { "POLYGON$it" }
|
||||
"MultiPolygon" -> wktWriter.write(geometry).removePrefix("MULTIPOLYGON ").let { "MULTIPOLYGON$it" }
|
||||
else -> wktWriter.write(geometry)
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val MIN_COORDINATE_VALUES = 2
|
||||
const val MIN_LINEAR_RING_POINTS = 4
|
||||
}
|
||||
}
|
||||
+200
@@ -0,0 +1,200 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.springframework.stereotype.Service
|
||||
import org.springframework.transaction.annotation.Transactional
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportDto
|
||||
import space.nstart.pcp.pcp_types_lib.message.KafkaMessage
|
||||
import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportEntity
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
|
||||
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository
|
||||
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportRepository
|
||||
import tools.jackson.databind.JsonNode
|
||||
import tools.jackson.databind.ObjectMapper
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
/**
|
||||
* Сохраняет разобранный route passport и outbox-события в одной DB transaction.
|
||||
*/
|
||||
@Service
|
||||
class RoutePassportIngestionService(
|
||||
private val routePassportRepository: RoutePassportRepository,
|
||||
private val routePassportOutboxRepository: RoutePassportOutboxRepository,
|
||||
private val objectMapper: ObjectMapper,
|
||||
private val kafkaTopicsProperties: KafkaTopicsProperties,
|
||||
) {
|
||||
|
||||
@Transactional
|
||||
fun ingest(
|
||||
parsedMessage: ParsedRoutePassportMessage,
|
||||
rawPayload: String,
|
||||
sourceRecord: RoutePassportSourceRecordMetadata,
|
||||
) {
|
||||
val now = LocalDateTime.now()
|
||||
val routePassportDto = parsedMessage.routePassportDto
|
||||
|
||||
val routePassportEntity = routePassportRepository.findById(routePassportDto.routeId)
|
||||
.map { existingEntity ->
|
||||
existingEntity.applyRoutePassport(
|
||||
routePassportDto = routePassportDto,
|
||||
parsedMessage = parsedMessage,
|
||||
rawPayload = rawPayload,
|
||||
sourceRecord = sourceRecord,
|
||||
updatedAt = now,
|
||||
)
|
||||
}
|
||||
.orElseGet {
|
||||
routePassportDto.toEntity(
|
||||
parsedMessage = parsedMessage,
|
||||
rawPayload = rawPayload,
|
||||
sourceRecord = sourceRecord,
|
||||
now = now,
|
||||
)
|
||||
}
|
||||
|
||||
routePassportRepository.save(routePassportEntity)
|
||||
|
||||
createOutboxIfAbsent(
|
||||
eventType = PcpKafkaEvent.ModeStatusChangedEvent,
|
||||
topic = kafkaTopicsProperties.processedRoute,
|
||||
parsedMessage = parsedMessage,
|
||||
now = now,
|
||||
)
|
||||
createOutboxIfAbsent(
|
||||
eventType = PcpKafkaEvent.RouteGeoRefEvent,
|
||||
topic = kafkaTopicsProperties.routeGeoreference,
|
||||
parsedMessage = parsedMessage,
|
||||
now = now,
|
||||
)
|
||||
}
|
||||
|
||||
private fun createOutboxIfAbsent(
|
||||
eventType: PcpKafkaEvent,
|
||||
topic: String,
|
||||
parsedMessage: ParsedRoutePassportMessage,
|
||||
now: LocalDateTime,
|
||||
) {
|
||||
val routeId = parsedMessage.routePassportDto.routeId
|
||||
val eventTypeName = eventType.name
|
||||
|
||||
if (routePassportOutboxRepository.existsByRouteIdAndEventType(routeId, eventTypeName)) {
|
||||
return
|
||||
}
|
||||
|
||||
routePassportOutboxRepository.save(
|
||||
RoutePassportOutboxEntity(
|
||||
id = UUID.randomUUID(),
|
||||
routeId = routeId,
|
||||
eventType = eventTypeName,
|
||||
topic = topic,
|
||||
messageKey = routeId.toString(),
|
||||
payload = buildOutboxPayload(eventType, parsedMessage),
|
||||
headers = null,
|
||||
status = RoutePassportOutboxStatus.PENDING,
|
||||
attempts = 0,
|
||||
maxAttempts = 10,
|
||||
nextAttemptAt = now,
|
||||
lockedUntil = null,
|
||||
lastError = null,
|
||||
createdAt = now,
|
||||
updatedAt = now,
|
||||
publishedAt = null,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
private fun buildOutboxPayload(
|
||||
eventType: PcpKafkaEvent,
|
||||
parsedMessage: ParsedRoutePassportMessage,
|
||||
): JsonNode {
|
||||
val message = KafkaMessage(
|
||||
type = eventType,
|
||||
data = parsedMessage.routePassportDto,
|
||||
traceId = parsedMessage.metadata.traceId,
|
||||
correlationId = parsedMessage.metadata.correlationId,
|
||||
group = "pcp",
|
||||
).apply {
|
||||
source = "route-processing"
|
||||
}
|
||||
|
||||
return objectMapper.valueToTree(message)
|
||||
}
|
||||
|
||||
private fun RoutePassportEntity.applyRoutePassport(
|
||||
routePassportDto: RoutePassportDto,
|
||||
parsedMessage: ParsedRoutePassportMessage,
|
||||
rawPayload: String,
|
||||
sourceRecord: RoutePassportSourceRecordMetadata,
|
||||
updatedAt: LocalDateTime,
|
||||
): RoutePassportEntity {
|
||||
routeNameFull = routePassportDto.routeNameFull
|
||||
routeNameShort = routePassportDto.routeNameShort
|
||||
kaShort = routePassportDto.kaShort
|
||||
kaFull = routePassportDto.kaFull
|
||||
orbitNumber = routePassportDto.orbitNumber
|
||||
orbitState = routePassportDto.orbitState
|
||||
intervalBegin = routePassportDto.intervalBegin
|
||||
intervalEnd = routePassportDto.intervalEnd
|
||||
rollAngleMin = routePassportDto.rollAngle.min
|
||||
rollAngleMax = routePassportDto.rollAngle.max
|
||||
visirAngleMin = routePassportDto.visirAngle.min
|
||||
visirAngleMax = routePassportDto.visirAngle.max
|
||||
resolutionRange = routePassportDto.resolutionRange
|
||||
resolutionAzimuth = routePassportDto.resolutionAzimuth
|
||||
polarisation = objectMapper.valueToTree(routePassportDto.polarisation)
|
||||
processingLevel = routePassportDto.processingLevel
|
||||
geometryWkt = routePassportDto.geometry
|
||||
routeStatus = routePassportDto.routeStatus.name
|
||||
traceId = parsedMessage.metadata.traceId
|
||||
correlationId = parsedMessage.metadata.correlationId
|
||||
sourceTopic = sourceRecord.topic
|
||||
sourcePartition = sourceRecord.partition
|
||||
sourceOffset = sourceRecord.offset
|
||||
sourceKey = sourceRecord.key
|
||||
this.rawPayload = objectMapper.readTree(rawPayload)
|
||||
this.updatedAt = updatedAt
|
||||
|
||||
return this
|
||||
}
|
||||
|
||||
private fun RoutePassportDto.toEntity(
|
||||
parsedMessage: ParsedRoutePassportMessage,
|
||||
rawPayload: String,
|
||||
sourceRecord: RoutePassportSourceRecordMetadata,
|
||||
now: LocalDateTime,
|
||||
): RoutePassportEntity =
|
||||
RoutePassportEntity(
|
||||
routeId = routeId,
|
||||
routeNameFull = routeNameFull,
|
||||
routeNameShort = routeNameShort,
|
||||
kaShort = kaShort,
|
||||
kaFull = kaFull,
|
||||
orbitNumber = orbitNumber,
|
||||
orbitState = orbitState,
|
||||
intervalBegin = intervalBegin,
|
||||
intervalEnd = intervalEnd,
|
||||
rollAngleMin = rollAngle.min,
|
||||
rollAngleMax = rollAngle.max,
|
||||
visirAngleMin = visirAngle.min,
|
||||
visirAngleMax = visirAngle.max,
|
||||
resolutionRange = resolutionRange,
|
||||
resolutionAzimuth = resolutionAzimuth,
|
||||
polarisation = objectMapper.valueToTree(polarisation),
|
||||
processingLevel = processingLevel,
|
||||
geometryWkt = geometry,
|
||||
routeStatus = routeStatus.name,
|
||||
traceId = parsedMessage.metadata.traceId,
|
||||
correlationId = parsedMessage.metadata.correlationId,
|
||||
sourceTopic = sourceRecord.topic,
|
||||
sourcePartition = sourceRecord.partition,
|
||||
sourceOffset = sourceRecord.offset,
|
||||
sourceKey = sourceRecord.key,
|
||||
rawPayload = objectMapper.readTree(rawPayload),
|
||||
createdAt = now,
|
||||
updatedAt = now,
|
||||
)
|
||||
}
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.springframework.stereotype.Service
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.AngleRangeDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RawFeaturePassportJsonDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RawPassportEnvelopeDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportMessageMetadata
|
||||
import space.nstart.pcp.routepassportconsumer.exception.RoutePassportContractException
|
||||
import tools.jackson.core.JacksonException
|
||||
import tools.jackson.databind.ObjectMapper
|
||||
import java.math.BigDecimal
|
||||
import java.nio.charset.StandardCharsets
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
/**
|
||||
* Сервис, который парсит JSON паспорта маршрута в итоговый DTO.
|
||||
*/
|
||||
@Service
|
||||
class RoutePassportJsonParserService(
|
||||
|
||||
/** Jackson ObjectMapper, который выполняет базовую десериализацию JSON. */
|
||||
private val objectMapper: ObjectMapper,
|
||||
|
||||
/** Converter GeoJSON geometry входного паспорта маршрута в WKT. */
|
||||
private val routeGeometryConverter: RouteGeometryConverter,
|
||||
) {
|
||||
|
||||
/**
|
||||
* Парсит входной JSON и возвращает готовый DTO.
|
||||
*
|
||||
* @param json исходное Kafka-сообщение envelope-формата с полезной нагрузкой в поле `data`.
|
||||
* @return DTO с уже преобразованными полями.
|
||||
*/
|
||||
fun parse(json: String): ParsedRoutePassportMessage {
|
||||
val rawEnvelope: RawPassportEnvelopeDto = try {
|
||||
objectMapper.readValue(json, RawPassportEnvelopeDto::class.java)
|
||||
} catch (exception: JacksonException) {
|
||||
throw RoutePassportContractException("Некорректный JSON contract паспорта маршрута", exception)
|
||||
}
|
||||
val routePassportDto = rawEnvelope.data.toRoutePassportDto()
|
||||
|
||||
return ParsedRoutePassportMessage(
|
||||
metadata = RoutePassportMessageMetadata(
|
||||
traceId = rawEnvelope.traceId,
|
||||
correlationId = rawEnvelope.correlationId,
|
||||
),
|
||||
routePassportDto = routePassportDto,
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Преобразует feature из envelope-сообщения в бизнес DTO.
|
||||
*/
|
||||
private fun RawFeaturePassportJsonDto.toRoutePassportDto(): RoutePassportDto {
|
||||
val routePassportDto = RoutePassportDto(
|
||||
routeId = routeId ?: UUID.nameUUIDFromBytes(id.toByteArray(StandardCharsets.UTF_8)),
|
||||
kaShort = properties.missionId,
|
||||
kaFull = properties.platform,
|
||||
routeNameFull = id,
|
||||
routeNameShort = collection,
|
||||
orbitNumber = properties.orbitNumber.toLong(),
|
||||
orbitState = properties.orbitState,
|
||||
intervalBegin = LocalDateTime.parse(properties.intervalBegin),
|
||||
intervalEnd = LocalDateTime.parse(properties.intervalEnd),
|
||||
rollAngle = properties.rollAngle.toAngleRangeDto(fieldName = "rollAngle"),
|
||||
visirAngle = properties.visirAngle.toAngleRangeDto(fieldName = "visirAngle"),
|
||||
resolutionRange = properties.resolutionRange.toBigDecimal(),
|
||||
resolutionAzimuth = properties.resolutionAzimuth.toBigDecimal(),
|
||||
polarisation = properties.polarisation,
|
||||
processingLevel = properties.processingLevel,
|
||||
geometry = routeGeometryConverter.convert(geometry),
|
||||
)
|
||||
|
||||
return routePassportDto
|
||||
}
|
||||
|
||||
/**
|
||||
* Преобразует список чисел в DTO диапазона.
|
||||
*
|
||||
* @receiver список значений диапазона.
|
||||
* @param fieldName имя поля для текста ошибки.
|
||||
* @return DTO диапазона с минимальным и максимальным значением.
|
||||
*/
|
||||
private fun List<BigDecimal>.toAngleRangeDto(fieldName: String): AngleRangeDto {
|
||||
if (this.isEmpty()) {
|
||||
throw RoutePassportContractException("Поле $fieldName не должно быть пустым")
|
||||
}
|
||||
|
||||
/** Минимальное значение диапазона. */
|
||||
val minValue: BigDecimal = this.minOrNull()
|
||||
?: throw RoutePassportContractException("Не удалось определить минимум для поля $fieldName")
|
||||
|
||||
/** Максимальное значение диапазона. */
|
||||
val maxValue: BigDecimal = this.maxOrNull()
|
||||
?: throw RoutePassportContractException("Не удалось определить максимум для поля $fieldName")
|
||||
|
||||
/** DTO диапазона после вычисления минимума и максимума. */
|
||||
val angleRangeDto: AngleRangeDto = AngleRangeDto(
|
||||
min = minValue,
|
||||
max = maxValue,
|
||||
)
|
||||
|
||||
return angleRangeDto
|
||||
}
|
||||
}
|
||||
+113
@@ -0,0 +1,113 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.apache.kafka.clients.producer.ProducerRecord
|
||||
import org.slf4j.LoggerFactory
|
||||
import org.springframework.kafka.core.KafkaTemplate
|
||||
import org.springframework.kafka.support.SendResult
|
||||
import org.springframework.stereotype.Service
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage
|
||||
import space.nstart.pcp.pcp_types_lib.message.KafkaMessage
|
||||
import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
|
||||
import tools.jackson.databind.ObjectMapper
|
||||
import java.nio.charset.StandardCharsets
|
||||
import java.util.concurrent.CompletableFuture
|
||||
|
||||
/**
|
||||
* Сервис публикации обработанного маршрута в Kafka.
|
||||
*/
|
||||
@Service
|
||||
class RoutePassportKafkaPublisherService(
|
||||
/** ObjectMapper сериализует итоговый DTO в JSON перед отправкой. */
|
||||
private val objectMapper: ObjectMapper,
|
||||
|
||||
/** KafkaTemplate отправляет одно и то же сообщение в несколько topic'ов. */
|
||||
private val kafkaTemplate: KafkaTemplate<String, String>,
|
||||
|
||||
/** Набор Kafka topic'ов, в которые публикуется результат обработки. */
|
||||
private val kafkaTopicsProperties: KafkaTopicsProperties,
|
||||
) {
|
||||
private val log = LoggerFactory.getLogger(this::class.java)
|
||||
|
||||
/**
|
||||
* Публикует полный RoutePassportDto в оба выходных topic'а.
|
||||
*/
|
||||
fun publish(parsedMessage: ParsedRoutePassportMessage) {
|
||||
val routePassportDto = parsedMessage.routePassportDto
|
||||
|
||||
log.info(
|
||||
"Publishing route passport: routeNameFull={}, processedRouteTopic={}, routeGeoreferenceTopic={}",
|
||||
routePassportDto.routeNameFull,
|
||||
kafkaTopicsProperties.processedRoute,
|
||||
kafkaTopicsProperties.routeGeoreference,
|
||||
)
|
||||
|
||||
sendWithLogging(
|
||||
topic = kafkaTopicsProperties.processedRoute,
|
||||
eventType = PcpKafkaEvent.ModeStatusChangedEvent,
|
||||
payload = buildPayload(
|
||||
eventType = PcpKafkaEvent.ModeStatusChangedEvent,
|
||||
parsedMessage = parsedMessage,
|
||||
),
|
||||
traceId = parsedMessage.metadata.traceId,
|
||||
routeNameFull = routePassportDto.routeNameFull,
|
||||
)
|
||||
sendWithLogging(
|
||||
topic = kafkaTopicsProperties.routeGeoreference,
|
||||
eventType = PcpKafkaEvent.RouteGeoRefEvent,
|
||||
payload = buildPayload(
|
||||
eventType = PcpKafkaEvent.RouteGeoRefEvent,
|
||||
parsedMessage = parsedMessage,
|
||||
),
|
||||
traceId = parsedMessage.metadata.traceId,
|
||||
routeNameFull = routePassportDto.routeNameFull,
|
||||
)
|
||||
}
|
||||
|
||||
private fun buildPayload(eventType: PcpKafkaEvent, parsedMessage: ParsedRoutePassportMessage): String {
|
||||
val message = KafkaMessage(
|
||||
type = eventType,
|
||||
data = parsedMessage.routePassportDto,
|
||||
traceId = parsedMessage.metadata.traceId,
|
||||
correlationId = parsedMessage.metadata.correlationId,
|
||||
group = "pcp",
|
||||
).apply {
|
||||
source = "route-processing"
|
||||
}
|
||||
|
||||
return objectMapper.writeValueAsString(message)
|
||||
}
|
||||
|
||||
private fun sendWithLogging(
|
||||
topic: String,
|
||||
eventType: PcpKafkaEvent,
|
||||
payload: String,
|
||||
traceId: String,
|
||||
routeNameFull: String,
|
||||
) {
|
||||
val record = ProducerRecord(topic, traceId, payload)
|
||||
record.headers().add("type", eventType.name.toByteArray(StandardCharsets.UTF_8))
|
||||
|
||||
val future: CompletableFuture<SendResult<String, String>> = kafkaTemplate.send(record)
|
||||
future.whenComplete { result, exception ->
|
||||
if (exception == null) {
|
||||
log.info(
|
||||
"Kafka message published: routeNameFull={}, topic={}, type={}, partition={}, offset={}",
|
||||
routeNameFull,
|
||||
topic,
|
||||
eventType,
|
||||
result.recordMetadata.partition(),
|
||||
result.recordMetadata.offset(),
|
||||
)
|
||||
} else {
|
||||
log.error(
|
||||
"Failed to publish kafka message: routeNameFull={}, topic={}, type={}",
|
||||
routeNameFull,
|
||||
topic,
|
||||
eventType,
|
||||
exception,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+11
@@ -0,0 +1,11 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
/**
|
||||
* Metadata входной Kafka record, из которой был получен route passport.
|
||||
*/
|
||||
data class RoutePassportSourceRecordMetadata(
|
||||
val topic: String,
|
||||
val partition: Int,
|
||||
val offset: Long,
|
||||
val key: String?,
|
||||
)
|
||||
@@ -0,0 +1,28 @@
|
||||
spring:
|
||||
application:
|
||||
name: pcp-route-processing-service
|
||||
profiles:
|
||||
default: local
|
||||
config:
|
||||
import: "configserver:"
|
||||
cloud:
|
||||
config:
|
||||
uri: ${CONFIG_SERVER_URI:http://192.168.100.160:38888}
|
||||
fail-fast: ${CONFIG_SERVER_FAIL_FAST:true}
|
||||
profile: ${SPRING_CLOUD_CONFIG_PROFILE:${SPRING_PROFILES_ACTIVE:${spring.profiles.default}}}
|
||||
label: ${SPRING_CLOUD_CONFIG_LABEL:master}
|
||||
datasource:
|
||||
driver-class-name: org.postgresql.Driver
|
||||
url: ${PCP_ROUTE_PROCESSING_DATASOURCE_URL:jdbc:postgresql://${PCP_POSTGRES_HOST:localhost}:${PCP_POSTGRES_PORT:5432}/pcp_route_processing}
|
||||
username: ${PCP_ROUTE_PROCESSING_DATASOURCE_USERNAME:postgres}
|
||||
password: ${PCP_ROUTE_PROCESSING_DATASOURCE_PASSWORD:password}
|
||||
jpa:
|
||||
hibernate:
|
||||
ddl-auto: validate
|
||||
properties:
|
||||
hibernate:
|
||||
format_sql: true
|
||||
flyway:
|
||||
enabled: true
|
||||
baseline-on-migrate: true
|
||||
locations: classpath:db/migration
|
||||
+46
@@ -0,0 +1,46 @@
|
||||
CREATE TABLE route_passports (
|
||||
route_id UUID PRIMARY KEY,
|
||||
route_name_full TEXT NOT NULL,
|
||||
route_name_short VARCHAR(255) NOT NULL,
|
||||
ka_short VARCHAR(64) NOT NULL,
|
||||
ka_full VARCHAR(255) NOT NULL,
|
||||
orbit_number BIGINT NOT NULL,
|
||||
orbit_state VARCHAR(64) NOT NULL,
|
||||
interval_begin TIMESTAMP NOT NULL,
|
||||
interval_end TIMESTAMP NOT NULL,
|
||||
roll_angle_min NUMERIC NOT NULL,
|
||||
roll_angle_max NUMERIC NOT NULL,
|
||||
visir_angle_min NUMERIC NOT NULL,
|
||||
visir_angle_max NUMERIC NOT NULL,
|
||||
resolution_range NUMERIC NOT NULL,
|
||||
resolution_azimuth NUMERIC NOT NULL,
|
||||
polarisation JSONB NOT NULL,
|
||||
processing_level VARCHAR(64) NOT NULL,
|
||||
geometry_wkt TEXT NOT NULL,
|
||||
route_status VARCHAR(32) NOT NULL,
|
||||
trace_id VARCHAR(255),
|
||||
correlation_id VARCHAR(255),
|
||||
source_topic VARCHAR(255),
|
||||
source_partition INTEGER,
|
||||
source_offset BIGINT,
|
||||
source_key VARCHAR(255),
|
||||
raw_payload JSONB,
|
||||
created_at TIMESTAMP NOT NULL,
|
||||
updated_at TIMESTAMP NOT NULL,
|
||||
version BIGINT NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE INDEX idx_route_passports_interval
|
||||
ON route_passports (interval_begin, interval_end);
|
||||
|
||||
CREATE INDEX idx_route_passports_ka_orbit
|
||||
ON route_passports (ka_short, orbit_number);
|
||||
|
||||
CREATE INDEX idx_route_passports_status
|
||||
ON route_passports (route_status);
|
||||
|
||||
CREATE UNIQUE INDEX ux_route_passports_source_record
|
||||
ON route_passports (source_topic, source_partition, source_offset)
|
||||
WHERE source_topic IS NOT NULL
|
||||
AND source_partition IS NOT NULL
|
||||
AND source_offset IS NOT NULL;
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
CREATE TABLE route_passport_outbox (
|
||||
id UUID PRIMARY KEY,
|
||||
route_id UUID NOT NULL,
|
||||
event_type VARCHAR(128) NOT NULL,
|
||||
topic VARCHAR(255) NOT NULL,
|
||||
message_key VARCHAR(255) NOT NULL,
|
||||
payload JSONB NOT NULL,
|
||||
headers JSONB,
|
||||
status VARCHAR(32) NOT NULL,
|
||||
attempts INTEGER NOT NULL DEFAULT 0,
|
||||
max_attempts INTEGER NOT NULL DEFAULT 10,
|
||||
next_attempt_at TIMESTAMP NOT NULL,
|
||||
locked_until TIMESTAMP,
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMP NOT NULL,
|
||||
updated_at TIMESTAMP NOT NULL,
|
||||
published_at TIMESTAMP,
|
||||
|
||||
CONSTRAINT fk_route_passport_outbox_route
|
||||
FOREIGN KEY (route_id)
|
||||
REFERENCES route_passports (route_id)
|
||||
ON DELETE CASCADE,
|
||||
|
||||
CONSTRAINT ux_route_passport_outbox_route_event
|
||||
UNIQUE (route_id, event_type),
|
||||
|
||||
CONSTRAINT chk_route_passport_outbox_status
|
||||
CHECK (status IN ('PENDING', 'PUBLISHED', 'FAILED'))
|
||||
);
|
||||
|
||||
CREATE INDEX idx_route_passport_outbox_due
|
||||
ON route_passport_outbox (status, next_attempt_at, created_at);
|
||||
|
||||
CREATE INDEX idx_route_passport_outbox_locked_until
|
||||
ON route_passport_outbox (locked_until);
|
||||
|
||||
CREATE INDEX idx_route_passport_outbox_route_id
|
||||
ON route_passport_outbox (route_id);
|
||||
@@ -0,0 +1,207 @@
|
||||
{
|
||||
"id": "51b07d12-1823-4bd5-b55d-de02a5304c03",
|
||||
"traceid": "c0dd08ea-dd64-4633-8845-d1b163bdf506",
|
||||
"correlationid": "db605d5f-77e9-45b4-82cc-709837306257",
|
||||
"group": "stac_test",
|
||||
"source": "stac-writer-service",
|
||||
"type": "METADATA_NORMALIZED",
|
||||
"time": "2026-03-31T09:39:13.251462",
|
||||
"data": {
|
||||
"id": "S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE",
|
||||
"routeId": "51b07d12-1823-4bd5-b55d-de02a5304c03",
|
||||
"type": "Feature",
|
||||
"geometry": {
|
||||
"type": "Polygon",
|
||||
"coordinates": [
|
||||
[
|
||||
[
|
||||
67.675827,
|
||||
38.114815
|
||||
],
|
||||
[
|
||||
64.779434,
|
||||
38.514816
|
||||
],
|
||||
[
|
||||
65.094017,
|
||||
40.015545
|
||||
],
|
||||
[
|
||||
68.05294,
|
||||
39.616795
|
||||
],
|
||||
[
|
||||
67.675827,
|
||||
38.114815
|
||||
]
|
||||
]
|
||||
]
|
||||
},
|
||||
"bbox": [
|
||||
64.779434,
|
||||
40.015545,
|
||||
68.05294,
|
||||
38.114815
|
||||
],
|
||||
"properties": {
|
||||
"datetime": "2026-03-31T12:08:30.570573Z",
|
||||
"created": null,
|
||||
"updated": null,
|
||||
"platform": "sentinel-1a",
|
||||
"instruments": null,
|
||||
"processing:level": "GRD",
|
||||
"route:orbit_number": "59796",
|
||||
"missionId": "S1A",
|
||||
"sat:orbit_state": "Descending",
|
||||
"polarisation": [
|
||||
"VH",
|
||||
"VV"
|
||||
],
|
||||
"rollAngle": [
|
||||
30.07864486915724,
|
||||
30.09378936608302
|
||||
],
|
||||
"visirAngle": [
|
||||
30.63331560676386,
|
||||
46.07086702096005
|
||||
],
|
||||
"intervalBegin": "2025-06-25T01:38:54.481190",
|
||||
"intervalEnd": "2025-06-25T01:39:19.479675",
|
||||
"resolutionRange": "1.000000e+01",
|
||||
"resolutionAzimuth": "1.000000e+01"
|
||||
},
|
||||
"links": [
|
||||
{
|
||||
"rel": "self",
|
||||
"href": "http://app-opensearch:8082/collections/S1A-GRDH/items/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE",
|
||||
"type": "application/geo+json",
|
||||
"title": null
|
||||
},
|
||||
{
|
||||
"rel": "parent",
|
||||
"href": "http://app-opensearch:8082/collections/S1A-GRDH",
|
||||
"type": "application/json",
|
||||
"title": null
|
||||
},
|
||||
{
|
||||
"rel": "collection",
|
||||
"href": "http://app-opensearch:8082/collections/S1A-GRDH",
|
||||
"type": "application/json",
|
||||
"title": null
|
||||
},
|
||||
{
|
||||
"rel": "root",
|
||||
"href": "http://app-opensearch:8082",
|
||||
"type": "application/json",
|
||||
"title": null
|
||||
}
|
||||
],
|
||||
"assets": {
|
||||
"schema-calibration-vh": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/calibration-s1a-iw-grd-vh-20250625t013854-20250625t013919-059796-076d15-002.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120818Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=26988c6a3490f71382ae9e61ff0d7b59aca32ec652993a785d24e8dd81e70939",
|
||||
"title": "xml",
|
||||
"description": null,
|
||||
"type": "application/xml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"schema-calibration-vv": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/calibration-s1a-iw-grd-vv-20250625t013854-20250625t013919-059796-076d15-001.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120818Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=edd6957a0404ead07e7a31d0759db26b3a6ad33394247c177dc41a7abf5972d0",
|
||||
"title": "xml",
|
||||
"description": null,
|
||||
"type": "application/xml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"schema-noise-vh": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/noise-s1a-iw-grd-vh-20250625t013854-20250625t013919-059796-076d15-002.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120818Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=3efe46114ed6dfeac898963510d3ba9c8b537c803b1fa69a5a03d5b462a1665c",
|
||||
"title": "xml",
|
||||
"description": null,
|
||||
"type": "application/xml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"schema-noise-vv": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/noise-s1a-iw-grd-vv-20250625t013854-20250625t013919-059796-076d15-001.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120818Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=0ca91479ba4132dbf13c2637f97567eb29ee9ca818009eaf501072995fe784ca",
|
||||
"title": "xml",
|
||||
"description": null,
|
||||
"type": "application/xml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"schema-product-vh": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/s1a-iw-grd-vh-20250625t013854-20250625t013919-059796-076d15-002.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120819Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=39817f0b7610c0b20a23d0fae24e7183f0d3719ba6ab2f315c320c9c6530274f",
|
||||
"title": "xml",
|
||||
"description": null,
|
||||
"type": "application/xml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"schema-product-vv": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/s1a-iw-grd-vv-20250625t013854-20250625t013919-059796-076d15-001.xml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120819Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=85d28bc2bb94724b4db7df84121d4252b423fa429204e4b25e06b9d725f3aeee",
|
||||
"title": "xml",
|
||||
"description": null,
|
||||
"type": "application/xml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"safe-manifest": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/manifest.safe?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120819Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=dbb22d57d031a412e2b46273edf32cfcb7fbdae1fa1fcc74bdcaa39a5e086f53",
|
||||
"title": "safe",
|
||||
"description": null,
|
||||
"type": "application/xml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"vh": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/s1a-iw-grd-vh-20250625t013854-20250625t013919-059796-076d15-002_cog.tif?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120824Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=dbecc9e04e976751e855f53e9a9a8db5d6365a30ac7ec116f31b0936ee375648",
|
||||
"title": "tif",
|
||||
"description": null,
|
||||
"type": "image/tiff; application=geotiff; profile=cloud-optimized",
|
||||
"roles": [
|
||||
"visual"
|
||||
]
|
||||
},
|
||||
"vv": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/s1a-iw-grd-vv-20250625t013854-20250625t013919-059796-076d15-001_cog.tif?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120829Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=34cf1e5f5174b7cb5ca5d173ff699923819d5df28486436d972efa59ac9b1ddb",
|
||||
"title": "tif",
|
||||
"description": null,
|
||||
"type": "image/tiff; application=geotiff; profile=cloud-optimized",
|
||||
"roles": [
|
||||
"visual"
|
||||
]
|
||||
},
|
||||
"kml": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/map-overlay.kml?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120829Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=92f63b9f463041c4ed87b3cb508ba8c57fd2db8ea2d22026087ba57d2ef88f54",
|
||||
"title": "kml",
|
||||
"description": null,
|
||||
"type": "application/kml",
|
||||
"roles": [
|
||||
"metadata"
|
||||
]
|
||||
},
|
||||
"thumbnail": {
|
||||
"href": "http://192.168.100.160:9000/stac-data/S1A/S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE/quick-look.png?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=minioadmin%2F20260331%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20260331T120829Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=19bebb1e446add1d6e0898b4329d2a3e016f59087719ce2b51dff5c487835cb0",
|
||||
"title": "png",
|
||||
"description": null,
|
||||
"type": "image/png",
|
||||
"roles": [
|
||||
"thumbnail",
|
||||
"overview"
|
||||
]
|
||||
}
|
||||
},
|
||||
"collection": "S1A-GRDH",
|
||||
"stac_version": "1.0.0",
|
||||
"stac_extensions": null
|
||||
},
|
||||
"datacontenttype": "application/json",
|
||||
"specversion": "1.0"
|
||||
}
|
||||
+190
@@ -0,0 +1,190 @@
|
||||
package space.nstart.pcp.routepassportconsumer.config
|
||||
|
||||
import org.apache.kafka.clients.consumer.Consumer
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord
|
||||
import org.apache.kafka.clients.producer.ProducerRecord
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.mockito.ArgumentCaptor
|
||||
import org.mockito.ArgumentMatchers.any
|
||||
import org.mockito.Mockito.mock
|
||||
import org.mockito.Mockito.never
|
||||
import org.mockito.Mockito.times
|
||||
import org.mockito.Mockito.verify
|
||||
import org.mockito.Mockito.`when`
|
||||
import org.springframework.boot.autoconfigure.AutoConfigurations
|
||||
import org.springframework.boot.autoconfigure.context.ConfigurationPropertiesAutoConfiguration
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner
|
||||
import org.springframework.context.annotation.Bean
|
||||
import org.springframework.context.annotation.Configuration
|
||||
import org.springframework.kafka.core.KafkaTemplate
|
||||
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer
|
||||
import org.springframework.kafka.listener.DefaultErrorHandler
|
||||
import org.springframework.kafka.listener.MessageListenerContainer
|
||||
import org.springframework.kafka.support.KafkaHeaders
|
||||
import org.springframework.kafka.support.SendResult
|
||||
import space.nstart.pcp.routepassportconsumer.exception.RoutePassportContractException
|
||||
import java.nio.charset.StandardCharsets
|
||||
import java.util.concurrent.CompletableFuture
|
||||
|
||||
class KafkaConsumerConfigTest {
|
||||
|
||||
private val contextRunner = ApplicationContextRunner()
|
||||
.withConfiguration(AutoConfigurations.of(ConfigurationPropertiesAutoConfiguration::class.java))
|
||||
.withUserConfiguration(TestPropertiesConfiguration::class.java, KafkaConsumerConfig::class.java)
|
||||
.withPropertyValues(
|
||||
"spring.kafka.bootstrap-servers=localhost:9092",
|
||||
"spring.kafka.consumer.group-id=pcp-route-processing-service",
|
||||
"spring.kafka.consumer.auto-offset-reset=earliest",
|
||||
"app.kafka.topics.input=$INPUT_TOPIC",
|
||||
"app.kafka.topics.input-dlq=$INPUT_DLQ_TOPIC",
|
||||
"app.kafka.topics.processed-route=pcp.route.in.v1",
|
||||
"app.kafka.topics.route-georeference=pcp.route.georeference.v1",
|
||||
"app.route-passport.consumer.retry.max-attempts=3",
|
||||
"app.route-passport.consumer.retry.backoff-ms=0",
|
||||
"app.route-passport.consumer.retry.max-backoff-ms=0",
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `should create error handler and recoverer beans`() {
|
||||
contextRunner.run { context ->
|
||||
assertThat(context).hasSingleBean(DefaultErrorHandler::class.java)
|
||||
assertThat(context).hasSingleBean(DeadLetterPublishingRecoverer::class.java)
|
||||
|
||||
val retryProperties = context.getBean(RoutePassportConsumerRetryProperties::class.java)
|
||||
assertThat(retryProperties.maxAttempts).isEqualTo(3)
|
||||
assertThat(retryProperties.backoffMs).isZero()
|
||||
assertThat(retryProperties.maxBackoffMs).isZero()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `dead letter recoverer should publish original string record to configured input dlq topic`() {
|
||||
contextRunner.run { context ->
|
||||
val kafkaTemplate = context.kafkaTemplate()
|
||||
val recoverer = context.getBean(DeadLetterPublishingRecoverer::class.java)
|
||||
val record = consumerRecord()
|
||||
|
||||
recoverer.accept(record, RoutePassportContractException("payload contract is invalid"))
|
||||
|
||||
val publishedRecord = capturePublishedRecord(kafkaTemplate)
|
||||
assertThat(publishedRecord.topic()).isEqualTo(INPUT_DLQ_TOPIC)
|
||||
assertThat(publishedRecord.partition()).isEqualTo(record.partition())
|
||||
assertThat(publishedRecord.key()).isEqualTo(record.key())
|
||||
assertThat(publishedRecord.value()).isEqualTo(record.value())
|
||||
assertThat(publishedRecord.headers().lastStringHeader("pcp-service-name"))
|
||||
.isEqualTo("pcp-route-processing-service")
|
||||
assertThat(publishedRecord.headers().lastStringHeader("pcp-exception-class"))
|
||||
.isEqualTo(RoutePassportContractException::class.java.name)
|
||||
assertThat(publishedRecord.headers().lastStringHeader("pcp-exception-message"))
|
||||
.isEqualTo("payload contract is invalid")
|
||||
assertThat(publishedRecord.headers().lastStringHeader("pcp-original-topic")).isEqualTo(INPUT_TOPIC)
|
||||
assertThat(publishedRecord.headers().lastStringHeader("pcp-original-partition")).isEqualTo("3")
|
||||
assertThat(publishedRecord.headers().lastStringHeader("pcp-original-offset")).isEqualTo("42")
|
||||
assertThat(publishedRecord.headers().lastStringHeader(KafkaHeaders.DLT_ORIGINAL_TOPIC))
|
||||
.isEqualTo(INPUT_TOPIC)
|
||||
assertThat(publishedRecord.headers().lastStringHeader(KafkaHeaders.DLT_EXCEPTION_FQCN))
|
||||
.isEqualTo(RoutePassportContractException::class.java.name)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `route passport contract exception should be recovered without retry`() {
|
||||
contextRunner.run { context ->
|
||||
val kafkaTemplate = context.kafkaTemplate()
|
||||
val errorHandler = context.getBean(DefaultErrorHandler::class.java)
|
||||
|
||||
val handled = errorHandler.handleOne(
|
||||
RoutePassportContractException("payload contract is invalid"),
|
||||
consumerRecord(),
|
||||
mockConsumer(),
|
||||
mock(MessageListenerContainer::class.java),
|
||||
)
|
||||
|
||||
assertThat(handled).isTrue()
|
||||
verify(kafkaTemplate, times(1)).send(anyProducerRecord())
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `runtime exception should be retried with bounded policy before dlq recovery`() {
|
||||
contextRunner.run { context ->
|
||||
val kafkaTemplate = context.kafkaTemplate()
|
||||
val errorHandler = context.getBean(DefaultErrorHandler::class.java)
|
||||
val record = consumerRecord()
|
||||
val consumer = mockConsumer()
|
||||
val container = mock(MessageListenerContainer::class.java)
|
||||
val exception = IllegalStateException("database is temporarily unavailable")
|
||||
|
||||
assertThat(errorHandler.handleOne(exception, record, consumer, container)).isFalse()
|
||||
assertThat(errorHandler.handleOne(exception, record, consumer, container)).isFalse()
|
||||
verify(kafkaTemplate, never()).send(anyProducerRecord())
|
||||
|
||||
assertThat(errorHandler.handleOne(exception, record, consumer, container)).isTrue()
|
||||
verify(kafkaTemplate, times(1)).send(anyProducerRecord())
|
||||
}
|
||||
}
|
||||
|
||||
private fun consumerRecord(): ConsumerRecord<String, String> =
|
||||
ConsumerRecord(
|
||||
INPUT_TOPIC,
|
||||
3,
|
||||
42L,
|
||||
"route-key",
|
||||
"""{"id":"route-1"}""",
|
||||
)
|
||||
|
||||
private fun capturePublishedRecord(
|
||||
kafkaTemplate: KafkaTemplate<String, String>,
|
||||
): ProducerRecord<String, String> {
|
||||
val captor = producerRecordCaptor()
|
||||
verify(kafkaTemplate, times(1)).send(captor.capture())
|
||||
return captor.value
|
||||
}
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
private fun producerRecordCaptor(): ArgumentCaptor<ProducerRecord<String, String>> =
|
||||
ArgumentCaptor.forClass(ProducerRecord::class.java) as ArgumentCaptor<ProducerRecord<String, String>>
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
private fun mockConsumer(): Consumer<String, String> =
|
||||
mock(Consumer::class.java) as Consumer<String, String>
|
||||
|
||||
private fun org.apache.kafka.common.header.Headers.lastStringHeader(name: String): String? =
|
||||
lastHeader(name)?.value()?.toString(StandardCharsets.UTF_8)
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableConfigurationProperties(
|
||||
KafkaTopicsProperties::class,
|
||||
RoutePassportConsumerRetryProperties::class,
|
||||
)
|
||||
private class TestPropertiesConfiguration {
|
||||
|
||||
@Bean
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
fun kafkaTemplate(): KafkaTemplate<String, String> {
|
||||
val kafkaTemplate = mock(KafkaTemplate::class.java) as KafkaTemplate<String, String>
|
||||
val sendResult = mock(SendResult::class.java) as SendResult<String, String>
|
||||
|
||||
`when`(kafkaTemplate.send(anyProducerRecord()))
|
||||
.thenReturn(CompletableFuture.completedFuture(sendResult))
|
||||
|
||||
return kafkaTemplate
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val INPUT_TOPIC = "pcp.request.survey-georeference.v1"
|
||||
const val INPUT_DLQ_TOPIC = "pcp.request.survey-georeference-dlq.v1"
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
fun anyProducerRecord(): ProducerRecord<String, String> =
|
||||
(any(ProducerRecord::class.java) as ProducerRecord<String, String>?)
|
||||
?: ProducerRecord("stub-topic", "stub-key", "stub-payload")
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
fun org.springframework.context.ApplicationContext.kafkaTemplate(): KafkaTemplate<String, String> =
|
||||
getBean(KafkaTemplate::class.java) as KafkaTemplate<String, String>
|
||||
}
|
||||
}
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
package space.nstart.pcp.routepassportconsumer.controller
|
||||
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.mockito.ArgumentCaptor
|
||||
import org.mockito.Mockito.mock
|
||||
import org.mockito.Mockito.times
|
||||
import org.mockito.Mockito.verify
|
||||
import org.springframework.kafka.core.KafkaTemplate
|
||||
import org.springframework.test.web.servlet.MockMvc
|
||||
import org.springframework.test.web.servlet.get
|
||||
import org.springframework.test.web.servlet.post
|
||||
import org.springframework.test.web.servlet.setup.MockMvcBuilders
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import java.nio.charset.StandardCharsets
|
||||
|
||||
class RouteProcessingTestControllerTest {
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
private val kafkaTemplate = mock(KafkaTemplate::class.java) as KafkaTemplate<String, String>
|
||||
private val kafkaTopicsProperties = KafkaTopicsProperties(
|
||||
input = "pcp.request.survey-georeference.v1",
|
||||
processedRoute = "pcp.route.in.v1",
|
||||
routeGeoreference = "pcp.route.georeference.v1",
|
||||
)
|
||||
|
||||
private val mockMvc: MockMvc = MockMvcBuilders
|
||||
.standaloneSetup(RouteProcessingTestController(kafkaTemplate, kafkaTopicsProperties))
|
||||
.build()
|
||||
|
||||
@Test
|
||||
fun `GET test endpoint returns service status`() {
|
||||
mockMvc.get("/api/route-processing/test")
|
||||
.andExpect {
|
||||
status { isOk() }
|
||||
jsonPath("$.status") { value("ok") }
|
||||
jsonPath("$.service") { value("pcp-route-processing-service") }
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `POST send endpoint publishes test passport message to input topic`() {
|
||||
mockMvc.post("/api/route-processing/test/send-passport-message")
|
||||
.andExpect {
|
||||
status { isOk() }
|
||||
jsonPath("$.status") { value("sent") }
|
||||
jsonPath("$.topic") { value("pcp.request.survey-georeference.v1") }
|
||||
jsonPath("$.resource") { value("classpath:test-passport-message.json") }
|
||||
}
|
||||
|
||||
val payloadCaptor = ArgumentCaptor.forClass(String::class.java)
|
||||
verify(kafkaTemplate, times(1)).send(
|
||||
org.mockito.Mockito.eq("pcp.request.survey-georeference.v1"),
|
||||
payloadCaptor.capture(),
|
||||
)
|
||||
|
||||
val expectedPayload = this::class.java.classLoader
|
||||
.getResourceAsStream("test-passport-message.json")
|
||||
?.readAllBytes()
|
||||
?.toString(StandardCharsets.UTF_8)
|
||||
?: error("Не найден файл test-passport-message.json в test resources")
|
||||
|
||||
kotlin.test.assertEquals(expectedPayload, payloadCaptor.value)
|
||||
}
|
||||
}
|
||||
+176
@@ -0,0 +1,176 @@
|
||||
package space.nstart.pcp.routepassportconsumer.listener
|
||||
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord
|
||||
import org.junit.jupiter.api.Assertions.assertThrows
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.mockito.Mockito.doThrow
|
||||
import org.mockito.Mockito.mock
|
||||
import org.mockito.Mockito.never
|
||||
import org.mockito.Mockito.times
|
||||
import org.mockito.Mockito.verify
|
||||
import org.mockito.Mockito.verifyNoInteractions
|
||||
import org.mockito.Mockito.`when`
|
||||
import org.springframework.kafka.support.Acknowledgment
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.AngleRangeDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportMessageMetadata
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import space.nstart.pcp.routepassportconsumer.service.RoutePassportIngestionService
|
||||
import space.nstart.pcp.routepassportconsumer.service.RoutePassportJsonParserService
|
||||
import space.nstart.pcp.routepassportconsumer.service.RoutePassportSourceRecordMetadata
|
||||
import java.math.BigDecimal
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
|
||||
class RoutePassportMessageListenerTest {
|
||||
|
||||
@Test
|
||||
fun `should parse incoming record, ingest it and acknowledge offset`() {
|
||||
val parserService = mock(RoutePassportJsonParserService::class.java)
|
||||
val ingestionService = mock(RoutePassportIngestionService::class.java)
|
||||
val acknowledgment = mock(Acknowledgment::class.java)
|
||||
val listener = RoutePassportMessageListener(
|
||||
routePassportJsonParserService = parserService,
|
||||
routePassportIngestionService = ingestionService,
|
||||
kafkaTopicsProperties = kafkaTopicsProperties(),
|
||||
)
|
||||
val rawPayload = """{"id":"route-1"}"""
|
||||
val record = consumerRecord(rawPayload)
|
||||
val parsedMessage = parsedMessage()
|
||||
val sourceRecord = sourceRecordMetadata(record)
|
||||
|
||||
`when`(parserService.parse(rawPayload)).thenReturn(parsedMessage)
|
||||
|
||||
listener.listen(record, acknowledgment)
|
||||
|
||||
verify(parserService).parse(rawPayload)
|
||||
verify(ingestionService).ingest(
|
||||
parsedMessage = parsedMessage,
|
||||
rawPayload = rawPayload,
|
||||
sourceRecord = sourceRecord,
|
||||
)
|
||||
verify(acknowledgment, times(1)).acknowledge()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should not ingest or acknowledge when parser throws exception`() {
|
||||
val parserService = mock(RoutePassportJsonParserService::class.java)
|
||||
val ingestionService = mock(RoutePassportIngestionService::class.java)
|
||||
val acknowledgment = mock(Acknowledgment::class.java)
|
||||
val listener = RoutePassportMessageListener(
|
||||
routePassportJsonParserService = parserService,
|
||||
routePassportIngestionService = ingestionService,
|
||||
kafkaTopicsProperties = kafkaTopicsProperties(),
|
||||
)
|
||||
val rawPayload = """{"id":"route-1"}"""
|
||||
val record = consumerRecord(rawPayload)
|
||||
val exception = IllegalArgumentException("payload is invalid")
|
||||
|
||||
`when`(parserService.parse(rawPayload)).thenThrow(exception)
|
||||
|
||||
assertThrows(IllegalArgumentException::class.java) {
|
||||
listener.listen(record, acknowledgment)
|
||||
}
|
||||
|
||||
verify(parserService).parse(rawPayload)
|
||||
verifyNoInteractions(ingestionService)
|
||||
verify(acknowledgment, never()).acknowledge()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should not acknowledge when ingestion throws exception`() {
|
||||
val parserService = mock(RoutePassportJsonParserService::class.java)
|
||||
val ingestionService = mock(RoutePassportIngestionService::class.java)
|
||||
val acknowledgment = mock(Acknowledgment::class.java)
|
||||
val listener = RoutePassportMessageListener(
|
||||
routePassportJsonParserService = parserService,
|
||||
routePassportIngestionService = ingestionService,
|
||||
kafkaTopicsProperties = kafkaTopicsProperties(),
|
||||
)
|
||||
val rawPayload = """{"id":"route-1"}"""
|
||||
val record = consumerRecord(rawPayload)
|
||||
val parsedMessage = parsedMessage()
|
||||
val sourceRecord = sourceRecordMetadata(record)
|
||||
val exception = IllegalStateException("database is unavailable")
|
||||
|
||||
`when`(parserService.parse(rawPayload)).thenReturn(parsedMessage)
|
||||
doThrow(exception).`when`(ingestionService).ingest(
|
||||
parsedMessage = parsedMessage,
|
||||
rawPayload = rawPayload,
|
||||
sourceRecord = sourceRecord,
|
||||
)
|
||||
|
||||
assertThrows(IllegalStateException::class.java) {
|
||||
listener.listen(record, acknowledgment)
|
||||
}
|
||||
|
||||
verify(parserService).parse(rawPayload)
|
||||
verify(ingestionService).ingest(
|
||||
parsedMessage = parsedMessage,
|
||||
rawPayload = rawPayload,
|
||||
sourceRecord = sourceRecord,
|
||||
)
|
||||
verify(acknowledgment, never()).acknowledge()
|
||||
}
|
||||
|
||||
private fun kafkaTopicsProperties(): KafkaTopicsProperties =
|
||||
KafkaTopicsProperties(
|
||||
input = INPUT_TOPIC,
|
||||
processedRoute = "pcp.route.in.v1",
|
||||
routeGeoreference = "pcp.route.georeference.v1",
|
||||
)
|
||||
|
||||
private fun consumerRecord(rawPayload: String): ConsumerRecord<String, String> =
|
||||
ConsumerRecord(
|
||||
INPUT_TOPIC,
|
||||
3,
|
||||
42L,
|
||||
"route-key",
|
||||
rawPayload,
|
||||
)
|
||||
|
||||
private fun sourceRecordMetadata(record: ConsumerRecord<String, String>): RoutePassportSourceRecordMetadata =
|
||||
RoutePassportSourceRecordMetadata(
|
||||
topic = record.topic(),
|
||||
partition = record.partition(),
|
||||
offset = record.offset(),
|
||||
key = record.key(),
|
||||
)
|
||||
|
||||
private fun parsedMessage(): ParsedRoutePassportMessage =
|
||||
ParsedRoutePassportMessage(
|
||||
metadata = RoutePassportMessageMetadata(
|
||||
traceId = "trace-1",
|
||||
correlationId = "corr-1",
|
||||
),
|
||||
routePassportDto = RoutePassportDto(
|
||||
routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03"),
|
||||
kaShort = "S1A",
|
||||
kaFull = "sentinel-1a",
|
||||
routeNameFull = "ROUTE-001",
|
||||
routeNameShort = "S1A-GRDH",
|
||||
orbitNumber = 59796L,
|
||||
orbitState = "Descending",
|
||||
intervalBegin = LocalDateTime.of(2026, 4, 1, 10, 15, 0),
|
||||
intervalEnd = LocalDateTime.of(2026, 4, 1, 10, 16, 30),
|
||||
rollAngle = AngleRangeDto(
|
||||
min = BigDecimal("-15.5"),
|
||||
max = BigDecimal("-12.0"),
|
||||
),
|
||||
visirAngle = AngleRangeDto(
|
||||
min = BigDecimal("21.0"),
|
||||
max = BigDecimal("24.5"),
|
||||
),
|
||||
resolutionRange = BigDecimal("10.0"),
|
||||
resolutionAzimuth = BigDecimal("10.0"),
|
||||
polarisation = listOf("VH", "VV"),
|
||||
processingLevel = "GRD",
|
||||
geometry = "POLYGON((0 0, 1 0, 1 1, 0 0))",
|
||||
),
|
||||
)
|
||||
|
||||
private companion object {
|
||||
const val INPUT_TOPIC = "pcp.request.survey-georeference.v1"
|
||||
}
|
||||
}
|
||||
+87
@@ -0,0 +1,87 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.junit.jupiter.api.Assertions.assertThrows
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RawGeoJsonGeometryDto
|
||||
import space.nstart.pcp.routepassportconsumer.exception.RoutePassportContractException
|
||||
import tools.jackson.databind.ObjectMapper
|
||||
import tools.jackson.databind.json.JsonMapper
|
||||
import tools.jackson.module.kotlin.KotlinModule
|
||||
|
||||
class RouteGeometryConverterTest {
|
||||
private val objectMapper: ObjectMapper = JsonMapper.builder()
|
||||
.addModule(KotlinModule.Builder().build())
|
||||
.build()
|
||||
|
||||
private val converter = RouteGeometryConverter()
|
||||
|
||||
@Test
|
||||
fun `should convert valid polygon to wkt`() {
|
||||
val geometry = geometry(
|
||||
type = "Polygon",
|
||||
coordinates = """[[[0, 0], [1, 0], [1, 1], [0, 0]]]""",
|
||||
)
|
||||
|
||||
val wkt = converter.convert(geometry)
|
||||
|
||||
assertTrue(wkt.startsWith("POLYGON"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should convert valid multipolygon to wkt`() {
|
||||
val geometry = geometry(
|
||||
type = "MultiPolygon",
|
||||
coordinates = """[[[[0, 0], [1, 0], [1, 1], [0, 0]]]]""",
|
||||
)
|
||||
|
||||
val wkt = converter.convert(geometry)
|
||||
|
||||
assertTrue(wkt.startsWith("MULTIPOLYGON"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should reject unsupported geometry type`() {
|
||||
val geometry = geometry(
|
||||
type = "Point",
|
||||
coordinates = """[0, 0]""",
|
||||
)
|
||||
|
||||
assertThrows(RoutePassportContractException::class.java) {
|
||||
converter.convert(geometry)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should reject malformed coordinates`() {
|
||||
val geometry = geometry(
|
||||
type = "Polygon",
|
||||
coordinates = """[[[0, 0], [1, 0], [1, 1]]]""",
|
||||
)
|
||||
|
||||
assertThrows(RoutePassportContractException::class.java) {
|
||||
converter.convert(geometry)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should reject non numeric coordinate values`() {
|
||||
val geometry = geometry(
|
||||
type = "Polygon",
|
||||
coordinates = """[[[0, 0], ["bad", 0], [1, 1], [0, 0]]]""",
|
||||
)
|
||||
|
||||
assertThrows(RoutePassportContractException::class.java) {
|
||||
converter.convert(geometry)
|
||||
}
|
||||
}
|
||||
|
||||
private fun geometry(
|
||||
type: String,
|
||||
coordinates: String,
|
||||
): RawGeoJsonGeometryDto =
|
||||
RawGeoJsonGeometryDto(
|
||||
type = type,
|
||||
coordinates = objectMapper.readTree(coordinates),
|
||||
)
|
||||
}
|
||||
+302
@@ -0,0 +1,302 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertNotNull
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.mockito.ArgumentMatchers.any
|
||||
import org.mockito.Mockito.doAnswer
|
||||
import org.mockito.Mockito.doReturn
|
||||
import org.mockito.Mockito.mock
|
||||
import org.mockito.Mockito.never
|
||||
import org.mockito.Mockito.verify
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.AngleRangeDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportMessageMetadata
|
||||
import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportEntity
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
|
||||
import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
|
||||
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository
|
||||
import space.nstart.pcp.routepassportconsumer.repository.RoutePassportRepository
|
||||
import tools.jackson.databind.ObjectMapper
|
||||
import tools.jackson.databind.json.JsonMapper
|
||||
import tools.jackson.module.kotlin.KotlinModule
|
||||
import java.math.BigDecimal
|
||||
import java.time.LocalDateTime
|
||||
import java.util.Optional
|
||||
import java.util.UUID
|
||||
|
||||
class RoutePassportIngestionServiceTest {
|
||||
private val objectMapper: ObjectMapper = JsonMapper.builder()
|
||||
.addModule(KotlinModule.Builder().build())
|
||||
.build()
|
||||
|
||||
private val kafkaTopicsProperties = KafkaTopicsProperties(
|
||||
input = "pcp.request.survey-georeference.v1",
|
||||
processedRoute = "pcp.route.in.v1",
|
||||
routeGeoreference = "pcp.route.georeference.v1",
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `first ingest creates route passport and two pending outbox events`() {
|
||||
val fixture = fixture()
|
||||
val routePassportRepository = mock(RoutePassportRepository::class.java)
|
||||
val routePassportOutboxRepository = mock(RoutePassportOutboxRepository::class.java)
|
||||
val savedRoutes = mutableListOf<RoutePassportEntity>()
|
||||
val savedOutbox = mutableListOf<RoutePassportOutboxEntity>()
|
||||
val service = service(routePassportRepository, routePassportOutboxRepository)
|
||||
|
||||
doReturn(Optional.empty<RoutePassportEntity>())
|
||||
.`when`(routePassportRepository)
|
||||
.findById(fixture.routeId)
|
||||
captureRouteSaves(routePassportRepository, savedRoutes)
|
||||
captureOutboxSaves(routePassportOutboxRepository, savedOutbox)
|
||||
|
||||
service.ingest(fixture.parsedMessage, fixture.rawPayload, fixture.sourceRecord)
|
||||
|
||||
assertEquals(1, savedRoutes.size)
|
||||
assertEquals(fixture.routeId, savedRoutes.single().routeId)
|
||||
assertEquals("ROUTE-001", savedRoutes.single().routeNameFull)
|
||||
assertTrue(savedRoutes.single().polarisation.isArray)
|
||||
assertEquals("""["VH","VV"]""", savedRoutes.single().polarisation.toString())
|
||||
assertTrue(savedRoutes.single().rawPayload?.isObject == true)
|
||||
assertEquals("PROCESSED", savedRoutes.single().routeStatus)
|
||||
|
||||
assertEquals(2, savedOutbox.size)
|
||||
assertEquals(
|
||||
listOf(PcpKafkaEvent.ModeStatusChangedEvent.name, PcpKafkaEvent.RouteGeoRefEvent.name),
|
||||
savedOutbox.map { it.eventType },
|
||||
)
|
||||
assertEquals(
|
||||
listOf("pcp.route.in.v1", "pcp.route.georeference.v1"),
|
||||
savedOutbox.map { it.topic },
|
||||
)
|
||||
savedOutbox.forEach { outbox ->
|
||||
assertEquals(fixture.routeId, outbox.routeId)
|
||||
assertEquals(fixture.routeId.toString(), outbox.messageKey)
|
||||
assertEquals(RoutePassportOutboxStatus.PENDING, outbox.status)
|
||||
assertEquals(0, outbox.attempts)
|
||||
assertEquals(10, outbox.maxAttempts)
|
||||
assertNotNull(outbox.nextAttemptAt)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `duplicate ingest updates route passport without creating duplicate outbox rows`() {
|
||||
val fixture = fixture()
|
||||
val existingEntity = existingRouteEntity(fixture.routeId).apply {
|
||||
createdAt = LocalDateTime.of(2026, 1, 1, 1, 0)
|
||||
updatedAt = LocalDateTime.of(2026, 1, 1, 1, 0)
|
||||
routeNameFull = "OLD-ROUTE"
|
||||
}
|
||||
val routePassportRepository = mock(RoutePassportRepository::class.java)
|
||||
val routePassportOutboxRepository = mock(RoutePassportOutboxRepository::class.java)
|
||||
val savedRoutes = mutableListOf<RoutePassportEntity>()
|
||||
val service = service(routePassportRepository, routePassportOutboxRepository)
|
||||
|
||||
doReturn(Optional.of(existingEntity))
|
||||
.`when`(routePassportRepository)
|
||||
.findById(fixture.routeId)
|
||||
doReturn(true)
|
||||
.`when`(routePassportOutboxRepository)
|
||||
.existsByRouteIdAndEventType(fixture.routeId, PcpKafkaEvent.ModeStatusChangedEvent.name)
|
||||
doReturn(true)
|
||||
.`when`(routePassportOutboxRepository)
|
||||
.existsByRouteIdAndEventType(fixture.routeId, PcpKafkaEvent.RouteGeoRefEvent.name)
|
||||
captureRouteSaves(routePassportRepository, savedRoutes)
|
||||
|
||||
service.ingest(fixture.parsedMessage, fixture.rawPayload, fixture.sourceRecord)
|
||||
|
||||
assertEquals(1, savedRoutes.size)
|
||||
assertEquals("ROUTE-001", existingEntity.routeNameFull)
|
||||
assertEquals(LocalDateTime.of(2026, 1, 1, 1, 0), existingEntity.createdAt)
|
||||
assertTrue(existingEntity.updatedAt.isAfter(existingEntity.createdAt))
|
||||
verify(routePassportOutboxRepository, never()).save(anyRoutePassportOutboxEntity())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ingest stores source record metadata in route passport`() {
|
||||
val fixture = fixture()
|
||||
val routePassportRepository = mock(RoutePassportRepository::class.java)
|
||||
val routePassportOutboxRepository = mock(RoutePassportOutboxRepository::class.java)
|
||||
val savedRoutes = mutableListOf<RoutePassportEntity>()
|
||||
val service = service(routePassportRepository, routePassportOutboxRepository)
|
||||
|
||||
doReturn(Optional.empty<RoutePassportEntity>())
|
||||
.`when`(routePassportRepository)
|
||||
.findById(fixture.routeId)
|
||||
captureRouteSaves(routePassportRepository, savedRoutes)
|
||||
|
||||
service.ingest(fixture.parsedMessage, fixture.rawPayload, fixture.sourceRecord)
|
||||
|
||||
val routeEntity = savedRoutes.single()
|
||||
assertEquals("pcp.request.survey-georeference.v1", routeEntity.sourceTopic)
|
||||
assertEquals(3, routeEntity.sourcePartition)
|
||||
assertEquals(42L, routeEntity.sourceOffset)
|
||||
assertEquals("source-key-1", routeEntity.sourceKey)
|
||||
assertEquals(objectMapper.readTree(fixture.rawPayload), routeEntity.rawPayload)
|
||||
assertEquals("trace-1", routeEntity.traceId)
|
||||
assertEquals("corr-1", routeEntity.correlationId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `outbox payload keeps downstream kafka message contract`() {
|
||||
val fixture = fixture()
|
||||
val routePassportRepository = mock(RoutePassportRepository::class.java)
|
||||
val routePassportOutboxRepository = mock(RoutePassportOutboxRepository::class.java)
|
||||
val savedOutbox = mutableListOf<RoutePassportOutboxEntity>()
|
||||
val service = service(routePassportRepository, routePassportOutboxRepository)
|
||||
|
||||
doReturn(Optional.empty<RoutePassportEntity>())
|
||||
.`when`(routePassportRepository)
|
||||
.findById(fixture.routeId)
|
||||
captureRouteSaves(routePassportRepository, mutableListOf())
|
||||
captureOutboxSaves(routePassportOutboxRepository, savedOutbox)
|
||||
|
||||
service.ingest(fixture.parsedMessage, fixture.rawPayload, fixture.sourceRecord)
|
||||
|
||||
val modeStatusPayload = savedOutbox.first().payload
|
||||
assertTrue(modeStatusPayload.isObject)
|
||||
assertEquals(PcpKafkaEvent.ModeStatusChangedEvent.name, modeStatusPayload.path("type").stringValue())
|
||||
assertEquals("trace-1", modeStatusPayload.path("traceId").stringValue())
|
||||
assertEquals("corr-1", modeStatusPayload.path("correlationId").stringValue())
|
||||
assertEquals("route-processing", modeStatusPayload.path("source").stringValue())
|
||||
assertEquals("pcp", modeStatusPayload.path("group").stringValue())
|
||||
assertTrue(modeStatusPayload.path("data").isObject)
|
||||
assertEquals(fixture.routeId.toString(), modeStatusPayload.path("data").path("routeId").stringValue())
|
||||
assertEquals("ROUTE-001", modeStatusPayload.path("data").path("routeNameFull").stringValue())
|
||||
}
|
||||
|
||||
private fun service(
|
||||
routePassportRepository: RoutePassportRepository,
|
||||
routePassportOutboxRepository: RoutePassportOutboxRepository,
|
||||
): RoutePassportIngestionService =
|
||||
RoutePassportIngestionService(
|
||||
routePassportRepository = routePassportRepository,
|
||||
routePassportOutboxRepository = routePassportOutboxRepository,
|
||||
objectMapper = objectMapper,
|
||||
kafkaTopicsProperties = kafkaTopicsProperties,
|
||||
)
|
||||
|
||||
private fun captureRouteSaves(
|
||||
routePassportRepository: RoutePassportRepository,
|
||||
savedRoutes: MutableList<RoutePassportEntity>,
|
||||
) {
|
||||
doAnswer { invocation ->
|
||||
val entity = invocation.arguments[0] as RoutePassportEntity
|
||||
savedRoutes += entity
|
||||
entity
|
||||
}.`when`(routePassportRepository).save(anyRoutePassportEntity())
|
||||
}
|
||||
|
||||
private fun captureOutboxSaves(
|
||||
routePassportOutboxRepository: RoutePassportOutboxRepository,
|
||||
savedOutbox: MutableList<RoutePassportOutboxEntity>,
|
||||
) {
|
||||
doAnswer { invocation ->
|
||||
val entity = invocation.arguments[0] as RoutePassportOutboxEntity
|
||||
savedOutbox += entity
|
||||
entity
|
||||
}.`when`(routePassportOutboxRepository).save(anyRoutePassportOutboxEntity())
|
||||
}
|
||||
|
||||
private fun anyRoutePassportEntity(): RoutePassportEntity =
|
||||
any(RoutePassportEntity::class.java) ?: existingRouteEntity(UUID.randomUUID())
|
||||
|
||||
private fun anyRoutePassportOutboxEntity(): RoutePassportOutboxEntity =
|
||||
any(RoutePassportOutboxEntity::class.java) ?: RoutePassportOutboxEntity(
|
||||
id = UUID.randomUUID(),
|
||||
routeId = UUID.randomUUID(),
|
||||
eventType = PcpKafkaEvent.ModeStatusChangedEvent.name,
|
||||
topic = "topic",
|
||||
messageKey = "key",
|
||||
payload = objectMapper.readTree("{}"),
|
||||
status = RoutePassportOutboxStatus.PENDING,
|
||||
attempts = 0,
|
||||
maxAttempts = 10,
|
||||
nextAttemptAt = LocalDateTime.now(),
|
||||
createdAt = LocalDateTime.now(),
|
||||
updatedAt = LocalDateTime.now(),
|
||||
)
|
||||
|
||||
private fun fixture(): IngestionFixture {
|
||||
val routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03")
|
||||
val routePassportDto = RoutePassportDto(
|
||||
routeId = routeId,
|
||||
kaShort = "S1A",
|
||||
kaFull = "sentinel-1a",
|
||||
routeNameFull = "ROUTE-001",
|
||||
routeNameShort = "S1A-GRDH",
|
||||
orbitNumber = 59796L,
|
||||
orbitState = "Descending",
|
||||
intervalBegin = LocalDateTime.of(2026, 4, 1, 10, 15, 0),
|
||||
intervalEnd = LocalDateTime.of(2026, 4, 1, 10, 16, 30),
|
||||
rollAngle = AngleRangeDto(
|
||||
min = BigDecimal("-15.5"),
|
||||
max = BigDecimal("-12.0"),
|
||||
),
|
||||
visirAngle = AngleRangeDto(
|
||||
min = BigDecimal("21.0"),
|
||||
max = BigDecimal("24.5"),
|
||||
),
|
||||
resolutionRange = BigDecimal("10.0"),
|
||||
resolutionAzimuth = BigDecimal("10.0"),
|
||||
polarisation = listOf("VH", "VV"),
|
||||
processingLevel = "GRD",
|
||||
geometry = "POLYGON((0 0, 1 0, 1 1, 0 0))",
|
||||
)
|
||||
|
||||
return IngestionFixture(
|
||||
routeId = routeId,
|
||||
parsedMessage = ParsedRoutePassportMessage(
|
||||
metadata = RoutePassportMessageMetadata(
|
||||
traceId = "trace-1",
|
||||
correlationId = "corr-1",
|
||||
),
|
||||
routePassportDto = routePassportDto,
|
||||
),
|
||||
rawPayload = """{"data":{"id":"ROUTE-001"}}""",
|
||||
sourceRecord = RoutePassportSourceRecordMetadata(
|
||||
topic = "pcp.request.survey-georeference.v1",
|
||||
partition = 3,
|
||||
offset = 42L,
|
||||
key = "source-key-1",
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
private fun existingRouteEntity(routeId: UUID): RoutePassportEntity =
|
||||
RoutePassportEntity(
|
||||
routeId = routeId,
|
||||
routeNameFull = "ROUTE-OLD",
|
||||
routeNameShort = "OLD",
|
||||
kaShort = "S1A",
|
||||
kaFull = "sentinel-1a",
|
||||
orbitNumber = 1L,
|
||||
orbitState = "Descending",
|
||||
intervalBegin = LocalDateTime.of(2026, 1, 1, 0, 0),
|
||||
intervalEnd = LocalDateTime.of(2026, 1, 1, 0, 1),
|
||||
rollAngleMin = BigDecimal.ZERO,
|
||||
rollAngleMax = BigDecimal.ONE,
|
||||
visirAngleMin = BigDecimal.ZERO,
|
||||
visirAngleMax = BigDecimal.ONE,
|
||||
resolutionRange = BigDecimal.TEN,
|
||||
resolutionAzimuth = BigDecimal.TEN,
|
||||
polarisation = objectMapper.readTree("""["VH"]"""),
|
||||
processingLevel = "GRD",
|
||||
geometryWkt = "POLYGON((0 0, 1 0, 1 1, 0 0))",
|
||||
routeStatus = "PROCESSED",
|
||||
createdAt = LocalDateTime.of(2026, 1, 1, 0, 0),
|
||||
updatedAt = LocalDateTime.of(2026, 1, 1, 0, 0),
|
||||
)
|
||||
|
||||
private data class IngestionFixture(
|
||||
val routeId: UUID,
|
||||
val parsedMessage: ParsedRoutePassportMessage,
|
||||
val rawPayload: String,
|
||||
val sourceRecord: RoutePassportSourceRecordMetadata,
|
||||
)
|
||||
}
|
||||
+198
@@ -0,0 +1,198 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertThrows
|
||||
import org.junit.jupiter.api.Test
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RawGeoJsonGeometryDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.SurveyModeStatus
|
||||
import space.nstart.pcp.routepassportconsumer.exception.RoutePassportContractException
|
||||
import java.util.UUID
|
||||
import tools.jackson.databind.ObjectMapper
|
||||
import tools.jackson.databind.json.JsonMapper
|
||||
import tools.jackson.module.kotlin.KotlinModule
|
||||
import java.nio.charset.StandardCharsets
|
||||
|
||||
/**
|
||||
* Тест сервиса парсинга JSON паспорта маршрута.
|
||||
*/
|
||||
class RoutePassportJsonParserServiceTest {
|
||||
private val objectMapper: ObjectMapper = JsonMapper.builder()
|
||||
.addModule(KotlinModule.Builder().build())
|
||||
.build()
|
||||
|
||||
@Test
|
||||
fun `should parse envelope passport json into dto`() {
|
||||
val parserService = RoutePassportJsonParserService(objectMapper, RouteGeometryConverter())
|
||||
|
||||
val json: String = this::class.java.classLoader
|
||||
.getResourceAsStream("test-passport-message.json")
|
||||
?.readAllBytes()
|
||||
?.toString(StandardCharsets.UTF_8)
|
||||
?: error("Не найден файл test-passport-message.json в classpath")
|
||||
|
||||
val parsedMessage = parserService.parse(json)
|
||||
val dto = parsedMessage.routePassportDto
|
||||
|
||||
assertEquals("c0dd08ea-dd64-4633-8845-d1b163bdf506", parsedMessage.metadata.traceId)
|
||||
assertEquals("db605d5f-77e9-45b4-82cc-709837306257", parsedMessage.metadata.correlationId)
|
||||
assertEquals(UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03"), dto.routeId)
|
||||
assertEquals("S1A", dto.kaShort)
|
||||
assertEquals("sentinel-1a", dto.kaFull)
|
||||
assertEquals("S1A-GRDH", dto.routeNameShort)
|
||||
assertEquals(
|
||||
"S1A_IW_GRDH_1SDV_20250625T013854_20250625T013919_059796_076D15_B5C1.SAFE",
|
||||
dto.routeNameFull,
|
||||
)
|
||||
assertEquals(59796L, dto.orbitNumber)
|
||||
assertEquals("Descending", dto.orbitState)
|
||||
assertEquals("GRD", dto.processingLevel)
|
||||
assertEquals(SurveyModeStatus.PROCESSED, dto.routeStatus)
|
||||
assertEquals(listOf("VH", "VV"), dto.polarisation)
|
||||
assertEquals(
|
||||
"POLYGON((67.675827 38.114815, 64.779434 38.514816, 65.094017 40.015545, 68.05294 39.616795, 67.675827 38.114815))",
|
||||
dto.geometry,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should generate route id when raw payload does not contain it`() {
|
||||
val parserService = RoutePassportJsonParserService(objectMapper, RouteGeometryConverter())
|
||||
|
||||
val json = """
|
||||
{
|
||||
"traceid": "trace-1",
|
||||
"correlationid": "corr-1",
|
||||
"data": {
|
||||
"id": "ROUTE-001",
|
||||
"collection": "S1A-GRDH",
|
||||
"geometry": {
|
||||
"type": "Polygon",
|
||||
"coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]
|
||||
},
|
||||
"properties": {
|
||||
"platform": "sentinel-1a",
|
||||
"missionId": "S1A",
|
||||
"route:orbit_number": "59796",
|
||||
"sat:orbit_state": "Descending",
|
||||
"intervalBegin": "2025-06-25T01:38:54.481190",
|
||||
"intervalEnd": "2025-06-25T01:39:19.479675",
|
||||
"rollAngle": [-5.0, 12.0],
|
||||
"visirAngle": [30.0, 46.0],
|
||||
"resolutionRange": "1.000000e+01",
|
||||
"resolutionAzimuth": "1.000000e+01",
|
||||
"polarisation": ["VH", "VV"],
|
||||
"processing:level": "GRD"
|
||||
}
|
||||
}
|
||||
}
|
||||
""".trimIndent()
|
||||
|
||||
val parsedMessage = parserService.parse(json)
|
||||
|
||||
assertEquals(
|
||||
UUID.nameUUIDFromBytes("ROUTE-001".toByteArray(StandardCharsets.UTF_8)),
|
||||
parsedMessage.routePassportDto.routeId,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should parse trace_id alias without changing correlationid`() {
|
||||
val parserService = RoutePassportJsonParserService(objectMapper, RouteGeometryConverter())
|
||||
|
||||
val json = """
|
||||
{
|
||||
"trace_id": "trace-aliased",
|
||||
"correlationid": "corr-aliased",
|
||||
"data": {
|
||||
"id": "ROUTE-001",
|
||||
"collection": "S1A-GRDH",
|
||||
"geometry": {
|
||||
"type": "Polygon",
|
||||
"coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]
|
||||
},
|
||||
"properties": {
|
||||
"platform": "sentinel-1a",
|
||||
"missionId": "S1A",
|
||||
"route:orbit_number": "59796",
|
||||
"sat:orbit_state": "Descending",
|
||||
"intervalBegin": "2025-06-25T01:38:54.481190",
|
||||
"intervalEnd": "2025-06-25T01:39:19.479675",
|
||||
"rollAngle": [-5.0, 12.0],
|
||||
"visirAngle": [30.0, 46.0],
|
||||
"resolutionRange": "1.000000e+01",
|
||||
"resolutionAzimuth": "1.000000e+01",
|
||||
"polarisation": ["VH", "VV"],
|
||||
"processing:level": "GRD"
|
||||
}
|
||||
}
|
||||
}
|
||||
""".trimIndent()
|
||||
|
||||
val parsedMessage = parserService.parse(json)
|
||||
|
||||
assertEquals("trace-aliased", parsedMessage.metadata.traceId)
|
||||
assertEquals("corr-aliased", parsedMessage.metadata.correlationId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should take geometry from converter`() {
|
||||
val routeGeometryConverter = StubRouteGeometryConverter()
|
||||
val parserService = RoutePassportJsonParserService(objectMapper, routeGeometryConverter)
|
||||
|
||||
val parsedMessage = parserService.parse(minimalJson("Polygon", "[[[0, 0], [1, 0], [1, 1], [0, 0]]]"))
|
||||
|
||||
assertEquals("POLYGON((9 9, 10 9, 10 10, 9 9))", parsedMessage.routePassportDto.geometry)
|
||||
assertEquals(1, routeGeometryConverter.callCount)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should reject unsupported geometry type in parser path`() {
|
||||
val parserService = RoutePassportJsonParserService(objectMapper, RouteGeometryConverter())
|
||||
|
||||
assertThrows(RoutePassportContractException::class.java) {
|
||||
parserService.parse(minimalJson("Point", "[0, 0]"))
|
||||
}
|
||||
}
|
||||
|
||||
private fun minimalJson(
|
||||
geometryType: String,
|
||||
coordinates: String,
|
||||
): String =
|
||||
"""
|
||||
{
|
||||
"traceid": "trace-1",
|
||||
"correlationid": "corr-1",
|
||||
"data": {
|
||||
"id": "ROUTE-001",
|
||||
"collection": "S1A-GRDH",
|
||||
"geometry": {
|
||||
"type": "$geometryType",
|
||||
"coordinates": $coordinates
|
||||
},
|
||||
"properties": {
|
||||
"platform": "sentinel-1a",
|
||||
"missionId": "S1A",
|
||||
"route:orbit_number": "59796",
|
||||
"sat:orbit_state": "Descending",
|
||||
"intervalBegin": "2025-06-25T01:38:54.481190",
|
||||
"intervalEnd": "2025-06-25T01:39:19.479675",
|
||||
"rollAngle": [-5.0, 12.0],
|
||||
"visirAngle": [30.0, 46.0],
|
||||
"resolutionRange": "1.000000e+01",
|
||||
"resolutionAzimuth": "1.000000e+01",
|
||||
"polarisation": ["VH", "VV"],
|
||||
"processing:level": "GRD"
|
||||
}
|
||||
}
|
||||
}
|
||||
""".trimIndent()
|
||||
|
||||
private class StubRouteGeometryConverter : RouteGeometryConverter() {
|
||||
var callCount: Int = 0
|
||||
|
||||
override fun convert(geometry: RawGeoJsonGeometryDto?): String {
|
||||
callCount += 1
|
||||
return "POLYGON((9 9, 10 9, 10 10, 9 9))"
|
||||
}
|
||||
}
|
||||
}
|
||||
+144
@@ -0,0 +1,144 @@
|
||||
package space.nstart.pcp.routepassportconsumer.service
|
||||
|
||||
import org.apache.kafka.clients.producer.ProducerRecord
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.mockito.ArgumentCaptor
|
||||
import org.mockito.ArgumentMatchers.any
|
||||
import org.mockito.Mockito.doAnswer
|
||||
import org.mockito.Mockito.doReturn
|
||||
import org.mockito.Mockito.mock
|
||||
import org.mockito.Mockito.times
|
||||
import org.mockito.Mockito.verify
|
||||
import org.springframework.kafka.core.KafkaTemplate
|
||||
import org.springframework.kafka.support.SendResult
|
||||
import space.nstart.pcp.routepassportconsumer.config.KafkaTopicsProperties
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.AngleRangeDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.ParsedRoutePassportMessage
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportDto
|
||||
import space.nstart.pcp.pcp_types_lib.dto.routes.RoutePassportMessageMetadata
|
||||
import space.nstart.pcp.pcp_types_lib.message.KafkaMessage
|
||||
import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
|
||||
import java.math.BigDecimal
|
||||
import java.nio.charset.StandardCharsets
|
||||
import java.time.LocalDateTime
|
||||
import java.util.UUID
|
||||
import java.util.concurrent.CompletableFuture
|
||||
import tools.jackson.databind.ObjectMapper
|
||||
|
||||
class RoutePassportKafkaPublisherServiceTest {
|
||||
|
||||
private fun anyProducerRecord(): ProducerRecord<String, String> {
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
return (any(ProducerRecord::class.java) as ProducerRecord<String, String>?)
|
||||
?: ProducerRecord("stub-topic", "stub-key", "stub-payload")
|
||||
}
|
||||
|
||||
private fun captureProducerRecord(
|
||||
captor: ArgumentCaptor<ProducerRecord<String, String>>,
|
||||
): ProducerRecord<String, String> {
|
||||
return captor.capture() ?: ProducerRecord("stub-topic", "stub-key", "stub-payload")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should publish kafka message to both output topics with dedicated event types`() {
|
||||
val objectMapper = mock(ObjectMapper::class.java)
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
val kafkaTemplate = mock(KafkaTemplate::class.java) as KafkaTemplate<String, String>
|
||||
val kafkaTopicsProperties = KafkaTopicsProperties(
|
||||
input = "pcp.request.survey-georeference.v1",
|
||||
processedRoute = "pcp.route.in.v1",
|
||||
routeGeoreference = "pcp.route.georeference.v1",
|
||||
)
|
||||
val publisherService = RoutePassportKafkaPublisherService(
|
||||
objectMapper = objectMapper,
|
||||
kafkaTemplate = kafkaTemplate,
|
||||
kafkaTopicsProperties = kafkaTopicsProperties,
|
||||
)
|
||||
val parsedMessage = ParsedRoutePassportMessage(
|
||||
metadata = RoutePassportMessageMetadata(
|
||||
traceId = "trace-1",
|
||||
correlationId = "corr-1",
|
||||
),
|
||||
routePassportDto = RoutePassportDto(
|
||||
routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03"),
|
||||
kaShort = "S1A",
|
||||
kaFull = "sentinel-1a",
|
||||
routeNameFull = "ROUTE-001",
|
||||
routeNameShort = "S1A-GRDH",
|
||||
orbitNumber = 59796L,
|
||||
orbitState = "Descending",
|
||||
intervalBegin = LocalDateTime.of(2026, 4, 1, 10, 15, 0),
|
||||
intervalEnd = LocalDateTime.of(2026, 4, 1, 10, 16, 30),
|
||||
rollAngle = AngleRangeDto(
|
||||
min = BigDecimal("-15.5"),
|
||||
max = BigDecimal("-12.0"),
|
||||
),
|
||||
visirAngle = AngleRangeDto(
|
||||
min = BigDecimal("21.0"),
|
||||
max = BigDecimal("24.5"),
|
||||
),
|
||||
resolutionRange = BigDecimal("10.0"),
|
||||
resolutionAzimuth = BigDecimal("10.0"),
|
||||
polarisation = listOf("VH", "VV"),
|
||||
processingLevel = "GRD",
|
||||
geometry = "POLYGON((0 0, 1 0, 1 1, 0 0))",
|
||||
),
|
||||
)
|
||||
val sendResult = mock(SendResult::class.java) as SendResult<String, String>
|
||||
|
||||
doAnswer { invocation ->
|
||||
val message = invocation.arguments[0] as KafkaMessage<*>
|
||||
"""{"type":"${message.type.enumCast()?.name}","routeNameFull":"${(message.data as RoutePassportDto).routeNameFull}"}"""
|
||||
}.`when`(objectMapper).writeValueAsString(any())
|
||||
doReturn(CompletableFuture.completedFuture(sendResult))
|
||||
.`when`(kafkaTemplate)
|
||||
.send(anyProducerRecord())
|
||||
|
||||
publisherService.publish(parsedMessage)
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
val recordCaptor = ArgumentCaptor.forClass(ProducerRecord::class.java) as ArgumentCaptor<ProducerRecord<String, String>>
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
val messageCaptor = ArgumentCaptor.forClass(KafkaMessage::class.java) as ArgumentCaptor<KafkaMessage<RoutePassportDto>>
|
||||
|
||||
verify(objectMapper, times(2)).writeValueAsString(messageCaptor.capture())
|
||||
verify(kafkaTemplate, times(2)).send(captureProducerRecord(recordCaptor))
|
||||
|
||||
kotlin.test.assertEquals(
|
||||
listOf(PcpKafkaEvent.ModeStatusChangedEvent, PcpKafkaEvent.RouteGeoRefEvent),
|
||||
messageCaptor.allValues.map { it.type },
|
||||
)
|
||||
messageCaptor.allValues.forEach { message ->
|
||||
kotlin.test.assertEquals("trace-1", message.traceId)
|
||||
kotlin.test.assertEquals("corr-1", message.correlationId)
|
||||
kotlin.test.assertEquals("pcp", message.group)
|
||||
kotlin.test.assertEquals("route-processing", message.source)
|
||||
kotlin.test.assertEquals("application/json", message.dataContentType)
|
||||
kotlin.test.assertEquals("1.0", message.specVersion)
|
||||
kotlin.test.assertEquals(parsedMessage.routePassportDto, message.data)
|
||||
kotlin.test.assertTrue(message.id.isNotBlank())
|
||||
kotlin.test.assertNotNull(message.time)
|
||||
}
|
||||
kotlin.test.assertEquals(
|
||||
listOf("pcp.route.in.v1", "pcp.route.georeference.v1"),
|
||||
recordCaptor.allValues.map { it.topic() },
|
||||
)
|
||||
kotlin.test.assertEquals(
|
||||
listOf("trace-1", "trace-1"),
|
||||
recordCaptor.allValues.map { it.key() },
|
||||
)
|
||||
kotlin.test.assertEquals(
|
||||
listOf(
|
||||
"""{"type":"ModeStatusChangedEvent","routeNameFull":"ROUTE-001"}""",
|
||||
"""{"type":"RouteGeoRefEvent","routeNameFull":"ROUTE-001"}""",
|
||||
),
|
||||
recordCaptor.allValues.map { it.value() },
|
||||
)
|
||||
kotlin.test.assertEquals(
|
||||
listOf("ModeStatusChangedEvent", "RouteGeoRefEvent"),
|
||||
recordCaptor.allValues.map {
|
||||
it.headers().lastHeader("type").value().toString(StandardCharsets.UTF_8)
|
||||
},
|
||||
)
|
||||
}
|
||||
}
|
||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user