From f6da0cb7824edea0e24a1a34ebc5a1d15db8079b Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 6 Sep 2026 20:53:14 +0800 Subject: [PATCH] =?UTF-8?q?feat(runtime):=20U07=20=E7=94=9F=E5=91=BD?= =?UTF-8?q?=E5=91=A8=E6=9C=9F=E8=A3=85=E9=85=8D=20+=20stub=20=E8=A3=85?= =?UTF-8?q?=E9=85=8D=E5=86=92=E7=83=9F=EF=BC=88ACM2-10=20U07/U01=20?= =?UTF-8?q?=E9=81=97=E7=95=99=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PipelineLifecycle：ServerStartupEvent 后在专用单线程（msgx-pump/msgx-dispatcher，daemon） 拉起 Pump/Dispatcher 循环；停机 requestStop+interrupt+join；@Requires(msgx.pipeline.autostart=true) - PipelineProps.pipeline.autostart（默认 false，生产/测试默认不自动起循环，防泄漏线程） - stub 装配（infra/stub，@Requires(msgx.stubs=true)）：7 仓储内存实装 + StubXmlCodec（未实装 → CODEC_ERROR 走 FAILED 可重放路径）+ StubRedis + StubDeliveryPort + Holder Factory - 配置：application-dev.yml（stubs+autostart，/env、/beans 放开，无需 DB 即可 dev 跑通）； application-test.yml（stubs=true、autostart=false）；README 补 dev stub 运行说明 - PipelineSmokeTest（4）：等价 /beans 装配核验 + 收报→主泵领取→FAILED(CODEC_ERROR) + U11 重放 回 PENDING + Dispatcher flush 聚合（stub 用例前清内存状态防上下文复用串扰） - 全量测试 34 个通过 --- README.md | 5 +- .../msgexchange/nextgen/PipelineLifecycle.kt | 50 ++++ .../nextgen/config/PipelineProps.kt | 3 + .../nextgen/delivery/Dispatcher.kt | 5 + .../nextgen/infra/stub/StubAdapters.kt | 76 ++++++ .../nextgen/infra/stub/StubRepositories.kt | 234 ++++++++++++++++++ .../msgexchange/nextgen/processing/Pump.kt | 5 + src/main/resources/application-dev.yml | 25 +- src/main/resources/application.yml | 1 + .../msgexchange/nextgen/PipelineSmokeTest.kt | 108 ++++++++ src/test/resources/application-test.yml | 13 +- 11 files changed, 518 insertions(+), 7 deletions(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubAdapters.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubRepositories.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt diff --git a/README.md b/README.md index 9185ac6..654c4d6 100644 --- a/README.md +++ b/README.md @@ -60,11 +60,14 @@ REQ_TRACK / PUMP_JOB / FLIGHT_STATE），并假定 `CMINMSGS`（及其历史表 ```bash ./gradlew build # 需网络拉取依赖；内网环境见 gradle.properties 注释 -./gradlew test # 纯逻辑单测（identity / schd 聚合 / 配置绑定） +./gradlew test # 纯逻辑单测（identity / schd 聚合 / 配置绑定 / 管道语义） +MICRONAUT_ENVIRONMENTS=dev ./gradlew run # dev stub 冒烟：内存仓储 + 启动主泵/投递（无需 DB/Redis/Kafka） ``` > 注解处理：Kotlin 侧经 KSP（`kotlin-ksp` + `micronaut-inject-kotlin`）生成 Micronaut > BeanDefinition（U01）；若 build 产物缺少 `*$Definition` 类，先检查 KSP 是否生效。 +> dev/shadow 冒烟装配：`msgx.stubs=true`（内存仓储/适配层，见 infra/stub）+ +> `msgx.pipeline.autostart=true`（PipelineLifecycle 拉起专用线程，U07）；生产默认两者关闭。 ## 关联 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt new file mode 100644 index 0000000..6cc2330 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt @@ -0,0 +1,50 @@ +package com.gzzn.omms.msgexchange.nextgen + +import com.gzzn.omms.msgexchange.nextgen.delivery.Dispatcher +import com.gzzn.omms.msgexchange.nextgen.processing.Pump +import io.micronaut.context.annotation.Requires +import io.micronaut.runtime.event.annotation.EventListener +import io.micronaut.runtime.server.event.ServerStartupEvent +import jakarta.annotation.PreDestroy +import jakarta.inject.Singleton + +/** + * U07（T03+N29）：管道生命周期装配——服务启动（ServerStartupEvent）后，在各自专用单线程 + * （msgx-pump / msgx-dispatcher，不占用 Netty event loop）上拉起 Pump 与 Dispatcher 循环； + * 停机时 requestStop + interrupt + join。仅当 `msgx.pipeline.autostart=true` 时装配 + * （默认关：需要真实仓储或 msgx.stubs=true 才安全开启）。 + */ +@Requires(property = "msgx.pipeline.autostart", value = "true") +@Singleton +class PipelineLifecycle( + private val pump: Pump, + private val dispatcher: Dispatcher, +) { + private val threads = mutableListOf() + + @Volatile + private var started = false + + @EventListener + fun start(event: ServerStartupEvent) { + startIfNeeded() + } + + private fun startIfNeeded() { + if (started) return + started = true + threads += spawn("msgx-pump", pump::loop) + threads += spawn("msgx-dispatcher", dispatcher::loop) + } + + private fun spawn(name: String, body: () -> Unit): Thread = + Thread.ofPlatform().name(name).daemon(true).start { body() } + + @PreDestroy + fun stop() { + pump.stop() + dispatcher.stop() + threads.forEach { it.interrupt() } // 解除 Thread.sleep 阻塞，加速退出 + threads.forEach { runCatching { it.join(3000) } } + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt index dcc8cc4..35840c4 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt @@ -29,6 +29,9 @@ class PipelineProps { var backoffCapMs: Long = 60_000 var headDeadline: Duration = Duration.ofMinutes(10) // 最坏 HOL 上界（毒丸升级） + /** U07：启动即拉起 Pump/Dispatcher 循环（默认关——需要真实仓储或 msgx.stubs=true 才可安全开启）。 */ + var autostart: Boolean = false + /** N28：attempt ≤ 0（如 FAILED 未递增 attempts 的行）不得抛异常，取下界=首档退避。 */ fun backoffFor(attempt: Int): Long { val index = (attempt - 1).coerceAtLeast(0) diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt index 41c30e2..a30b550 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt @@ -43,6 +43,11 @@ class Dispatcher( private var lastFlush: Instant = Instant.EPOCH + /** 优雅停机：loop 收尾后退出；线程中断由 Runner 负责。 */ + fun stop() { + running = false + } + fun loop() { while (running) { try { diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubAdapters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubAdapters.kt new file mode 100644 index 0000000..7598342 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubAdapters.kt @@ -0,0 +1,76 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.stub + +import com.gzzn.omms.msgexchange.nextgen.codec.DecodeFailure +import com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult +import com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec +import com.gzzn.omms.msgexchange.nextgen.delivery.DeliveryPort +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient +import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript +import com.gzzn.omms.msgexchange.nextgen.processing.CodecHolder +import com.gzzn.omms.msgexchange.nextgen.processing.HandlerHolder +import com.gzzn.omms.msgexchange.nextgen.processing.HandlerRegistry +import io.micronaut.context.annotation.Bean +import io.micronaut.context.annotation.Factory +import io.micronaut.context.annotation.Requires +import jakarta.inject.Singleton + +/** + * U07：stub 适配层——XmlCodec / Redis 客户端 / DeliveryPort / Handler 装配，仅在 msgx.stubs=true 时生效。 + * stub codec 未实装 → 报文 decode 返回 CODEC_ERROR（走 FAILED 可重放路径，链路上可观测）， + * 语义见 ACM2-10 U11：不把“未实装”写成报文非法/终态。 + */ +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubXmlCodec : XmlCodec { + override fun decode(rawXml: String): DecodeResult = + DecodeResult.Err(DecodeFailure(ErrorClass.CODEC_ERROR, "stub:codec-not-implemented")) + + override fun encodeRqrd(kind: String, rangeJson: String): String = "" +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubRedis : FlightRedisClient { + val hash = mutableMapOf() + val evalCalls = mutableListOf() + + fun clear() { hash.clear(); evalCalls.clear() } + + override fun eval(script: RedisScript, setPairs: List>, delFields: List) { + evalCalls += script + when (script) { + RedisScript.SNAPSHOT_REPLACE -> { + hash.putAll(setPairs) + delFields.forEach { hash.remove(it) } + } + RedisScript.BATCH_DELETE -> delFields.forEach { hash.remove(it) } + } + } + + override fun hgetAllFlightInfo(): Map = hash.toMap() +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubDeliveryPort : DeliveryPort { + val sent = mutableListOf>() + + fun clear() { sent.clear() } + + override fun sendKafka(topic: String, payloadJson: String) { sent += topic to payloadJson } + override fun indexFlightHts(payloadJson: String) = Unit + override fun projectRedis(payloadJson: String) = Unit +} + +/** Holder 工厂：CodecHolder/HandlerHolder 由 Micronaut Bean 提供（取代直连构造占位）。 */ +@Factory +@Requires(property = "msgx.stubs", value = "true") +class StubHolderFactory { + + @Bean + fun codecHolder(codec: StubXmlCodec): CodecHolder = CodecHolder(codec) + + @Bean + fun handlerHolder(): HandlerHolder = HandlerHolder(HandlerRegistry(emptyList())) // 阶段 2 前无 Handler 注册 +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubRepositories.kt new file mode 100644 index 0000000..8043328 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/stub/StubRepositories.kt @@ -0,0 +1,234 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.stub + +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.ProcState +import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.FlightStateRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.PumpJobRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository +import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository +import io.micronaut.context.annotation.Requires +import jakarta.inject.Singleton +import java.time.Instant +import java.util.concurrent.atomic.AtomicLong + +/** + * U07（U01 遗留 stub）：内存仓储装配——仅当 `msgx.stubs=true`（dev/影子冒烟）时生效 + * （@Requires），让「启动 + 收报 + 主泵/投递循环 + /beans 端到端」在无 MySQL/Redis/Kafka 时可跑。 + * 与真实 Micronaut Data 实装按模块替换；切换点条件显式，不误入生产。 + */ +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubProcState : ProcStateRepository { + private val rows = linkedMapOf() + private val bound = mutableMapOf() + + fun clear() { rows.clear(); bound.clear() } + + override fun insert(cminmsgsId: Long, state: ProcStatus) { + rows[cminmsgsId] = ProcState(cminmsgsId, state) + } + + override fun headUnfinished(): ProcState? = + rows.filterValues { it.state == ProcStatus.PENDING || it.state == ProcStatus.FAILED } + .minByOrNull { it.key }?.value + + override fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean { + val owner = bound[identityKey] + if (owner != null && owner != cminmsgsId) return false + bound[identityKey] = cminmsgsId + rows[cminmsgsId] = (rows[cminmsgsId] ?: ProcState(cminmsgsId, ProcStatus.PENDING)).copy(identityKey = identityKey) + return true + } + + override fun ownerOfIdentity(identityKey: String): Long? = bound[identityKey] + + override fun update( + cminmsgsId: Long, state: ProcStatus, nextAttemptAt: Instant?, attempts: Int?, + errorClass: ErrorClass?, lastError: String?, + ) { + val old = rows[cminmsgsId] ?: ProcState(cminmsgsId, state) + rows[cminmsgsId] = old.copy( + state = state, + nextAttemptAt = nextAttemptAt ?: old.nextAttemptAt, + attempts = attempts ?: old.attempts, + errorClass = errorClass ?: old.errorClass, + lastError = lastError ?: old.lastError, + ) + } + + override fun requeueByErrorClasses(errorClasses: List): Int { + var n = 0 + rows.keys.toList().forEach { id -> + val s = rows[id]!! + if (s.errorClass != null && s.errorClass in errorClasses && + (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) { + rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null) + n++ + } + } + return n + } + + /** 测试/运维观测用：读取当前状态行。 */ + fun snapshotOf(cminmsgsId: Long): ProcState? = rows[cminmsgsId] +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubInbox : CminmsgInboxRepository { + private val raws = linkedMapOf() + + fun clear() { raws.clear() } + private val ids = AtomicLong(0) + + override fun insertRaw(rawXml: String): Long { + val id = ids.incrementAndGet() + raws[id] = rawXml + return id + } + + override fun rawOf(cminmsgsId: Long): String? = raws[cminmsgsId] + + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) { + // 内存 stub：无列可回填，记录即可（后续由真实实装/审计消费） + } +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubMsgEvents : MsgEventRepository { + private val rows = linkedMapOf() + + fun clear() { rows.clear() } + private val ids = AtomicLong(0) + + override fun insertAll(events: List): List = events.map { e -> + val id = ids.incrementAndGet() + rows[id] = e.copy(eventId = id) + id + } + + override fun headUnsent(target: String): MsgEvent? = + rows.values.filter { it.target == target && it.state == EventStatus.PENDING } + .minByOrNull { it.eventId ?: Long.MAX_VALUE } + + override fun claimBatch(target: String, limit: Int): List = + rows.values.filter { it.target == target && it.state == EventStatus.PENDING } + .sortedBy { it.eventId ?: Long.MAX_VALUE } + .take(limit) + + override fun markSent(eventId: Long) = mutate(eventId) { it.copy(state = EventStatus.SENT) } + + override fun markAllSent(eventIds: List) = eventIds.forEach(::markSent) + + override fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) = + mutate(eventId) { it.copy(state = EventStatus.PENDING, attempts = attempts, nextAttemptAt = nextAttemptAt) } + + override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int?) = + mutate(eventId) { it.copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError, attempts = attempts ?: it.attempts) } + + override fun insertSync(events: List) { + insertAll(events) + } + + private fun mutate(eventId: Long, f: (MsgEvent) -> MsgEvent) { + val cur = rows[eventId] ?: return + rows[eventId] = f(cur) + } +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubPumpJobs : PumpJobRepository { + + fun clear() { rows.clear() } + data class JobRow(val job: PumpJobRepository.Job, var state: String, var lastError: String? = null) + + private val rows = linkedMapOf() + private val ids = AtomicLong(0) + + override fun enqueue(kind: String) { + val id = ids.incrementAndGet() + rows[id] = JobRow(PumpJobRepository.Job(id, kind), "QUEUED") + } + + override fun headQueued(): PumpJobRepository.Job? = + rows.values.firstOrNull { it.state == "QUEUED" }?.job + + override fun markRunning(jobId: Long) { rows[jobId]?.state = "RUNNING" } + override fun markDone(jobId: Long) { rows[jobId]?.state = "DONE" } + override fun markFailed(jobId: Long, lastError: String) { rows[jobId]?.state = "FAILED"; rows[jobId]?.lastError = lastError } +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubRefData : RefDataRepository { + private val gens = mutableMapOf() + private val flat = mutableMapOf, String>() + + fun clear() { gens.clear(); flat.clear() } + + override fun getGen(day: String): RefDataRepository.GenMeta? = gens[day] + + override fun putGenIfVersion(day: String, expected: Long, new: RefDataRepository.GenMeta): Boolean { + val cur = gens[day]?.version ?: 0L + if (cur != expected) return false + gens[day] = new + return true + } + + override fun upsertAll(rows: List>) { + rows.forEach { flat[it.first to it.second] = it.third } + } +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubReqTrack : ReqTrackRepository { + private val rows = mutableMapOf() + + fun clear() { rows.clear() } + private val ids = AtomicLong(0) + + override fun findOpenByKind(kind: String): ReqTrackRepository.Req? = + rows.values.firstOrNull { it.reqType == kind && it.state in setOf("REGISTERED", "SENT", "WAITING") } + + override fun forceExpireOpenOf(kind: String) { + rows.keys.toList().forEach { id -> + val r = rows[id]!! + if (r.reqType == kind && r.state in setOf("REGISTERED", "SENT", "WAITING")) + rows[id] = r.copy(state = "EXPIRED") + } + } + + override fun insert(kind: String, paramsJson: String): Long { + val id = ids.incrementAndGet() + rows[id] = ReqTrackRepository.Req(id, kind, "REGISTERED") + return id + } + + override fun linkCoutmsgs(reqId: Long, coutmsgsId: Long) = Unit + override fun markSent(reqId: Long, sentAt: Instant) { rows[reqId]?.let { rows[reqId] = it.copy(state = "SENT", sentAt = sentAt) } } + override fun expireIfWaiting(reqId: Long) { rows[reqId]?.let { rows[reqId] = it.copy(state = "EXPIRED") } } + override fun markDone(reqId: Long) { rows[reqId]?.let { rows[reqId] = it.copy(state = "DONE") } } +} + +@Requires(property = "msgx.stubs", value = "true") +@Singleton +class StubFlightState : FlightStateRepository { + private val byDay = mutableMapOf>>() + + fun clear() { byDay.clear() } + + override fun replaceDay(day: String, flights: List>) { + byDay[day] = flights.toMutableList() + } + + override fun findByDay(day: String): List> = byDay[day] ?: emptyList() +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt index 41d1040..ad22324 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt @@ -34,6 +34,11 @@ class Pump( @Volatile private var running = true + /** 优雅停机：loop 在当前 tick 收尾后退出；线程中断由 Runner 负责。 */ + fun stop() { + running = false + } + fun loop() { while (running) { try { diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index a23075d..1a93891 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -1,6 +1,27 @@ -# 开发/影子对拍环境 profile（U03/N32）：放开管理端点可见性以便“改值—回读”与 /beans 装配核验。 -# 生产环境不加载本 profile（/env、/beans 保持默认 sensitive）。 +# 开发/影子对拍环境 profile（U03/N32 + U07）： +# - /env、/beans 放开以便“改值—回读”与装配核验（生产 profile 不加载本文件） +# - msgx.stubs=true：无 MySQL/Redis/Kafka 时以内存 stub 仓储跑通「启动 + 收报 + 主泵/投递循环」 +# - msgx.pipeline.autostart=true：启动即拉起 Pump/Dispatcher 专用线程 # 启动：MICRONAUT_ENVIRONMENTS=dev ./gradlew run +msgx: + register-eureka: false + stubs: true + pipeline: + autostart: true + +# DB 仅在真实实装接入后需要；stub 模式不建连（Flyway 关闭；datasource 保留默认值以免占位符解析失败） +flyway: + datasources: + default: + enabled: false + +datasources: + default: + url: ${MSGX_DB_URL:jdbc:mysql://127.0.0.1:3306/msgexchange_dev} + username: ${MSGX_DB_USER:root} + password: ${MSGX_DB_PASSWORD:} + driver-class-name: com.mysql.cj.jdbc.Driver + micronaut: endpoints: env: diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 517755f..818359b 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -15,6 +15,7 @@ msgx: backoff-ms: [1000, 2000, 4000, 8000, 16000] # 指数退避，单次封顶 60s backoff-cap-ms: 60000 head-deadline: 10m # 队头滞留上界 = 最坏 HOL 时长（毒丸升级） + autostart: false # U07：启动即拉起 Pump/Dispatcher 循环；需真实仓储或 msgx.stubs=true 才开启（dev 见 application-dev.yml） schd: flush-period: 3s # KEEP 现役推送节律 flush-limit: 500 # 批上限，防积压尖峰 diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt new file mode 100644 index 0000000..1f33085 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt @@ -0,0 +1,108 @@ +package com.gzzn.omms.msgexchange.nextgen + +import com.gzzn.omms.msgexchange.nextgen.delivery.Dispatcher +import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass +import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent +import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus +import com.gzzn.omms.msgexchange.nextgen.domain.Targets +import com.gzzn.omms.msgexchange.nextgen.infra.stub.StubDeliveryPort +import com.gzzn.omms.msgexchange.nextgen.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.nextgen.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.nextgen.ingress.InboxController +import com.gzzn.omms.msgexchange.nextgen.processing.Pump +import io.micronaut.context.ApplicationContext +import io.micronaut.test.extensions.junit5.annotation.MicronautTest +import jakarta.inject.Inject +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 + +/** + * U07 端到端（stub 装配，等价 /beans 核验）：无 MySQL/Redis/Kafka 下， + * ①核心 bean 装配齐全；②收报→落库→主泵领取→解码未实装→FAILED(CODEC_ERROR)+退避； + * ③U11 重放把 FAILED 拉回 PENDING；④Dispatcher flushSchd 聚合发出 schd。 + * 后台循环关闭（autostart=false），按需手动 tick，避免测试泄漏线程。 + */ +@MicronautTest +class PipelineSmokeTest { + + @Inject + lateinit var ctx: ApplicationContext + + @Inject + lateinit var controller: InboxController + + @Inject + lateinit var pump: Pump + + @Inject + lateinit var dispatcher: Dispatcher + + // @MicronautTest 复用同一上下文：用例前清空内存 stub，避免全局状态跨用例串扰 + @org.junit.jupiter.api.BeforeEach + fun cleanStubs() { + ctx.getBean(StubProcState::class.java).clear() + ctx.getBean(com.gzzn.omms.msgexchange.nextgen.infra.stub.StubInbox::class.java).clear() + ctx.getBean(StubMsgEvents::class.java).clear() + ctx.getBean(StubDeliveryPort::class.java).clear() + } + + @Test + fun `core pipeline beans are wired under stubs`() { + assertNotNull(ctx.getBean(com.gzzn.omms.msgexchange.nextgen.config.PipelineProps::class.java)) + assertTrue(ctx.getBeansOfType(Pump::class.java).isNotEmpty()) + assertTrue(ctx.getBeansOfType(Dispatcher::class.java).isNotEmpty()) + assertNotNull(controller) + } + + @Test + fun `accept then pump tick transitions message to FAILED CODEC_ERROR with backoff`() { + val receipt = controller.send("") + assertNotNull(receipt.body()) // 受理 ID + val id = receipt.body()!!.toLong() + + pump.tick() // 主泵领取 → stub codec 未实装 → FAILED(CODEC_ERROR) + + val stub = ctx.getBean(StubProcState::class.java) + val s = stub.snapshotOf(id) + assertNotNull(s) + assertEquals(ProcStatus.FAILED, s!!.state) + assertEquals(ErrorClass.CODEC_ERROR, s.errorClass) + assertEquals(1, s.attempts) + assertNotNull(s.nextAttemptAt) + } + + @Test + fun `replay reopens failed CODEC_ERROR row to PENDING`() { + val receipt = controller.send("") + val id = receipt.body()!!.toLong() + pump.tick() + val stub = ctx.getBean(StubProcState::class.java) + assertEquals(ProcStatus.FAILED, stub.snapshotOf(id)!!.state) + + val n = ctx.getBean(com.gzzn.omms.msgexchange.nextgen.infra.retry.ReplayService::class.java) + .replay(listOf(ErrorClass.CODEC_ERROR)) + + assertEquals(1, n) + val reopened = stub.snapshotOf(id)!! + assertEquals(ProcStatus.PENDING, reopened.state) + assertEquals(0, reopened.attempts) + } + + @Test + fun `dispatcher flushes schd batch through stub port`() { + val events = ctx.getBean(StubMsgEvents::class.java) + val port = ctx.getBean(StubDeliveryPort::class.java) + events.insertAll(listOf( + MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", payloadJson = """{"FLID":"F1","v":"old"}"""), + MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", payloadJson = """{"FLID":"F1","v":"new"}"""), + )) + + dispatcher.flushSchd() + + assertEquals(1, port.sent.count { it.first == "schd" }) + val payload = port.sent.first { it.first == "schd" }.second + assertTrue(payload.contains("new") && !payload.contains("old")) + } +} diff --git a/src/test/resources/application-test.yml b/src/test/resources/application-test.yml index f5281ad..0812b01 100644 --- a/src/test/resources/application-test.yml +++ b/src/test/resources/application-test.yml @@ -1,6 +1,14 @@ # U03 配置绑定回归测试专用环境（@MicronautTest 默认 environment=test） # 目的：让 ApplicationContext 可无外部依赖启动（内存 datasource / 关闭迁移 / 关闭注册）， -# 以便以「注入 PipelineProps 断言」离线验证 msgx.* 嵌套键绑定（N02 验收），不依赖 /env 端点。 +# 以便以「注入 PipelineProps/DataSource 断言」离线验证绑定（N02/R09/R01a），不依赖 /env 端点。 +# U07：stubs=true 提供内存仓储使全链路（Controller→Inbox→Pump→Dispatcher）可装配； +# autostart=false：测试内不自动拉起后台循环（避免泄漏线程），按需手动 tick。 +msgx: + register-eureka: false + stubs: true + pipeline: + autostart: false + datasources: default: url: jdbc:h2:mem:cfgtest;DB_CLOSE_DELAY=-1 @@ -20,9 +28,6 @@ kafka: bootstrap: servers: 127.0.0.1:9092 -msgx: - register-eureka: false - micronaut: server: port: 0