diff --git a/README.md b/README.md index 7a2ac5c..d515f30 100644 --- a/README.md +++ b/README.md @@ -1,26 +1,42 @@ -# msgexchange-v2(新一代消息交换服务) +# msgexchange-v2(机场上游报文处理中间件) + +**系统角色**:消费 CIIMS/AODB 等上游经共享 MySQL 信箱(`CMINMSGS`)投递的 XML 报文, +解析处理后维护 Redis 航班动态并向 Kafka / 出站信箱投递——**中间件**,非报文源系统。 依据 **ACMA-8 v4 综合架构**(单写者严格 FIFO 管道)与 **ACMA-6 技术选型** -(Micronaut 5.1 + Kotlin 2.3)搭建的新一代消息交换服务工程;**存储边界与阶段 B 按 -ACM2-11/ACM2-12 定案**(自有 PostgreSQL 全内部状态 + 共享 MySQL 仅信箱 + Redis 动态/gen, -阶段 B 缓做)。本仓库独立于 legacy `msgexchange-api`(Java 8 / Spring Boot 1.5 / Maven)—— +(Micronaut 5.1 + Kotlin 2.3)搭建;**存储边界与阶段 B 按 ACM2-11/ACM2-12 定案** +(自有 PostgreSQL 全内部状态 + 共享 MySQL 仅信箱 + Redis 动态/gen,阶段 B 缓做)。 +本仓库独立于 legacy `msgexchange-api`(Java 8 / Spring Boot 1.5 / Maven)—— 过渡期两套系统并存(影子对拍→切流→旧仓库冻结),legacy 维护不受本仓库影响。 -> **JDK 口径实测修正**:Micronaut 5.1 系构件(如 micronaut-http-server-netty:5.1.10) -> 要求 JVM 25+,计划原定 JDK 21 不可行;工程已按 **JDK 25** 配置(ACMA-9 记录)。 +> **JDK 口径实测修正**:Micronaut 5.1 系构件要求 **JVM 25+**(`./gradlew :dependencies` 实测 +> core 系解析 **5.1.13**,platform BOM **5.1.3**,classpath 混用);计划原定 JDK 21 不可行; +> 工程已按 **JDK 25** 配置(ACMA-9 记录)。 + +## 系统边界 + +| 方向 | 机制 | 说明 | +|---|---|---| +| **入站(主路径)** | JDBC 轮询共享 `CMINMSGS` | 上游经 CIIMS 等**外部系统**写信箱;本系统 1s 轮询 `DATE_PROCESSED IS NULL` 发现新信 → 自有 PG 入队(`InboxPoller`,U05) | +| **入站(compat)** | HTTP `POST /cminmsgs/send` | 手工注入/影子对拍;写信箱 + PG 入队,**非**生产主拓扑 | +| **处理** | 主泵 FIFO 管道 | 解码 → identity → Handler 决策 → Redis 动态权威态 | +| **出站** | Kafka + `COUTMSGS` | 向下游推送 msg/schd;请求类报文写出站信箱 | + +与 SIS / legacy 一致:本系统**不**替代 CIIMS 落信,**不**生成原始 AODB 业务报文。 ## 包结构 → ACMA-8 架构映射 | 包 | 职责 | 对应 ACMA-8 | |---|---|---| -| `ingress/` | Ingress & Inbox:收报(共享信箱落信 → 自有 PG 入队,ACM2-12),不解析报文 | 流程 1,I3 | +| `ingress/` | Ingress & Inbox:JDBC 轮询共享信箱发现新信 → 自有 PG 入队(+ HTTP compat 写路径),不解析报文 | 流程 1,I3 | | `processing/` | Processing 主泵:严格 FIFO 领取、identity 绑定、纯函数决策、事务2 | 流程 2/4,I1/I2/I5 | | `delivery/` | Delivery & Projection:每 target 严格 FIFO 投递、schd 聚合 | 流程 3 | | `reference/` | Reference & Query:21 类同步 + 15 类请求状态机 | 流程 6 | | `jobs/` | 泵作业:ARCHIVE / HISTORY_SWEEP / PROJECTION_REBUILD | 流程 4/5/7,I4 | | `codec/` | XML codec(阶段 1 先 vendor 复用 legacy POJO,见 ACMA-6 选型) | 决策 4 前置 | | `domain/` | 领域模型:Decision、事件、状态机枚举、Phase 开关 | I1–I5 | -| `infra/` | 仓储接口、Redis Lua 装载、配置 | 数据模型节 | +| `config/` | `PipelineProps` 参数表(ACMA-8 参数初值,`msgx.*`) | — | +| `infra/` | 仓储接口、Redis Lua 装载、stub、健康、重试策略 | 数据模型节 | ## 资源 @@ -43,11 +59,11 @@ ACM2-11/ACM2-12 定案**(自有 PostgreSQL 全内部状态 + 共享 MySQL 仅 阶段 1 后续项。 3. **依赖版本锁定**:`gradle/libs.versions.toml` 中版本为计划口径,需阶段 0 「Micronaut×现网 Eureka 互操作冒烟 + logstash + ES REST」通过后固化。 -4. **自有 PG 数据层实装**(ACM2-12):`infra/persistence/Repositories.kt` 目前是接口 - (Micronaut Data JDBC on PG + 信箱适配层 CminmsgMailbox/OutboxMailbox 属 U05 批次), - 主泵/调度循环以接口驱动,纯逻辑已抽离可单测。 -5. **信箱适配层与共享库边界**(ACM2-12):CMINMSGS 收 / COUTMSGS 出为共享 MySQL 仅 DML - (mailbox.shared-mysql 配置段已占位);收报「信箱落信 → PG 入队」与处理回填的 +4. **自有 PG 数据层 + InboxPoller**(ACM2-12):`infra/persistence/Repositories.kt` 目前是接口 + (Micronaut Data JDBC on PG + 信箱适配层 CminmsgMailbox/OutboxMailbox + JDBC 轮询入队 + 属 U05 批次),主泵/调度循环以接口驱动,纯逻辑已抽离可单测。 +5. **信箱适配层与共享库边界**(ACM2-12):上游外部写 `CMINMSGS`;本系统 JDBC 轮询读 + + 处理回填写;出站写 `COUTMSGS`(mailbox.shared-mysql 配置段已占位);入队/回填的 外部副作用/补偿模型属 U05 批次。 ## 数据库初始化(ACM2-12 口径) @@ -56,12 +72,13 @@ ACM2-11/ACM2-12 定案**(自有 PostgreSQL 全内部状态 + 共享 MySQL 仅 PROC_STATE / MSG_EVENT / PUMP_JOB / REQ_TRACK / REF_MASTER(PG 方言)。全新库直接执行即可, 无 legacy 前置。 -**共享 MySQL(cdairport,他人系统库)**:本系统**不建表/schema**,仅信箱 DML——收报写 -`CMINMSGS`、出站写 `COUTMSGS`(他人读取发送);表结构与保留策略归库方管理。收报前需确认 -共享库 CMINMSGS 已存在(他人系统提供);本仓库不声明其 schema。 +**共享 MySQL(cdairport,他人系统库)**:本系统**不建表/schema**,仅信箱 DML——上游外部写 +`CMINMSGS`;本系统 JDBC 轮询读 + 处理回填;出站写 `COUTMSGS`(他人读取发送);表结构与 +保留策略归库方管理。部署前需确认共享库 CMINMSGS 已存在(他人系统提供);本仓库不声明其 schema。 -- **事务模型**:与共享库交互均为外部副作用(ACM2-12)——收报=信箱落信成功(返回 - CMINMSGS_ID)→ 自有 PG 建 PENDING 入队;处理成功回填 DATE_PROCESSED/STATUS 为最终一致。 +- **事务模型**:与共享库交互均为外部副作用(ACM2-12)——主路径=上游外部写信箱 → + JDBC 轮询发现新信 → 自有 PG 建 PENDING 入队(失败重扫补建);HTTP `/cminmsgs/send` + 为 compat 写路径;处理成功回填 DATE_PROCESSED/STATUS 为最终一致。 - **快照 gen**:随航班动态同在 Redis(gen→Redis 协议属 U09,当前为过渡占位)。 - **阶段 B(FLIGHT_STATE)**:缓做不落表。 - 影子对拍:自有 PG 开独立 schema;共享信箱为单信箱无法双写,影子输入=只读水位/回放口径。 @@ -81,9 +98,11 @@ MICRONAUT_ENVIRONMENTS=dev ./gradlew run # dev stub 冒烟:内存 stub,无 > /health 拖成 500(dev 关闭 redis.health/kafka.health,健康由自定义指示器承担);管理端点前缀误用 > `micronaut.endpoints.*`(正确为顶层 `endpoints.*`,/env、/beans 这才真正放开);另关闭 eureka > discovery。当前 `MICRONAUT_ENVIRONMENTS=dev ./gradlew run` **无需任何环境变量/外部依赖**即可启动, -> `POST /cminmsgs/send`(application/json)返回记录 ID,`/health` UP。 -> 已知边界:text/plain 收报 415 属 U16 缺口(契约对拍后固化);测试侧端到端语义由 -> PipelineSmokeTest(内嵌上下文)覆盖。 +> compat HTTP `POST /cminmsgs/send`(application/json)返回记录 ID,`/health` UP。 +> 已知边界:text/plain compat 写 415 属 U16 缺口(契约对拍后固化);`application-dev.yml` 中 +> `datasources.default.url` 仍为 MySQL 占位(ACM2-12 前残留,stub 下 `enabled=false` 不建连, +> U05 清理为 PG 口径);测试侧端到端语义由 PipelineSmokeTest(内嵌上下文,compat 写路径)覆盖。 +> **生产主路径**(JDBC 轮询 `InboxPoller`)属 U05,dev stub 尚未覆盖。 > 注解处理:Kotlin 侧经 KSP(`kotlin-ksp` + `micronaut-inject-kotlin`)生成 Micronaut > BeanDefinition(U01);若 build 产物缺少 `*$Definition` 类,先检查 KSP 是否生效。 @@ -92,17 +111,17 @@ MICRONAUT_ENVIRONMENTS=dev ./gradlew run # dev stub 冒烟:内存 stub,无 ## 文档 -- [docs/architecture.md](docs/architecture.md):架构速览——总体拓扑、模块职责、关键决策、 - 数据边界、部署与安全姿态、可观测性、就绪度(与代码同步维护)。 -- [docs/design.md](docs/design.md):设计细节——状态机与错误分类、数据模型、核心流程语义、 - 失败/重试/重放统一设计、不变量落点、参数表、测试策略、已知缺口(对照 ACM2-10)。 +- [docs/architecture.md](docs/architecture.md):架构速览——**中间件定位**、总体拓扑(JDBC 轮询主路径)、 + 上下游边界、模块职责、关键决策、数据边界、部署与安全姿态、可观测性、就绪度。 +- [docs/design.md](docs/design.md):设计细节——系统边界、状态机与错误分类、数据模型、 + 核心流程语义(流程 1 主/compath 分述)、失败/重试/重放、不变量落点、参数表、已知缺口。 - [docs/legacy/](docs/legacy/):外部参考/基线材料(自 legacy 仓库拷贝,非本系统文档)—— - `msgexchange-api-legacy-user-stories.md`(行为对拍基线,ACMA-4)、`unisysaodbsis.xsd`、 - `SIS_AODB_RMS-V0.1.md`(消息结构唯一事实源)。 + `msgexchange-api-legacy-user-stories.md`(legacy 行为对拍基线,ACMA-4)、`unisysaodbsis.xsd`、 + `SIS_AODB_RMS-V0.1.md`(消息结构唯一事实源;CIIMS 中间件交换模型)。 ## 关联 -Plane `airport_chengdu_msgexchange_api`(**ACM2 为现行入口**):ACM2-3(综合架构 v4, -架构权威,存储边界按 ACM2-11/12 修订)、ACM2-4(脚手架跟踪)、ACM2-10(评审与实施计划 -U01–U30)、ACM2-11/12(存储边界与共享信箱决策);ACMA 系列仅作归档历史/迁移来源 +Plane **`airport_chengdu_msgexchange_v2`(ACM2,现行入口)**:ACM2-3(综合架构 v4, +架构权威,存储边界按 ACM2-12 修订;ACM2-11 为决策史)、ACM2-4(脚手架跟踪)、ACM2-10(评审与实施计划 +U01–U30)、ACM2-12(存储边界与共享信箱决策);ACMA 系列仅作归档历史/迁移来源 (ACMA-8 v4 / ACMA-6 选型 / ACMA-9 JDK 口径在归档中可溯)。 diff --git a/docs/architecture.md b/docs/architecture.md index 96c02f6..e1295f5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,6 +1,8 @@ # msgexchange-v2 架构文档 -> 现行架构权威为 Plane `airport_chengdu_msgexchange_api` 工作区的 **ACM2-3(综合架构 v4)**; +> **系统角色**:机场 OMMS **上游报文处理中间件**——消费 CIIMS/AODB 等上游写入共享信箱的 +> XML 报文,经 FIFO 管道处理后向下游投递;**非**报文源系统。 +> 现行架构权威为 Plane **`airport_chengdu_msgexchange_v2`(ACM2)** 工作区的 **ACM2-3(综合架构 v4)**; > 脚手架跟踪 **ACM2-4**,评审与实施计划(U01–U30)**ACM2-10**。本文是仓库内的架构速览, > 与代码同步维护;两者冲突时以 ACM2-3 为准并回改本文。 > **存储边界与阶段 B 以 ACM2-11/ACM2-12 决策为准**(自有 PostgreSQL + 共享 MySQL 信箱 + @@ -10,27 +12,41 @@ ## 1. 系统定位 -新一代机场消息交换服务(AODB 报文接入 → 处理 → 对外投递),替换 legacy -`msgexchange-api`(Java 8 / Spring Boot 1.5 / Maven)。过渡策略为**双跑三步**: +**机场 OMMS 上游报文处理中间件**:消费 CIIMS/AODB 等上游写入共享信箱的 XML 报文, +经严格 FIFO 管道解析、决策、维护航班动态权威态,并向下游(Kafka / 出站信箱 / 查询接口) +投递;**非**报文源系统,**不**替代 CIIMS 或 AODB。替换 legacy `msgexchange-api` +(Java 8 / Spring Boot 1.5 / Maven)。过渡策略为**双跑三步**: ``` -影子对拍(同入口双收,比对输出)→ 切流(nextgen 权威)→ 旧仓库冻结 +影子对拍(共享库水位/回放 + 自有 PG 独立 schema 比对)→ 切流(nextgen 权威)→ 旧仓库冻结 ``` - legacy 维护不受本仓库影响;本仓库不声明 legacy 旧表 schema,**且不在共享 MySQL 建任何表** (CMINMSGS/COUTMSGS 所在库属他人系统,本系统仅信箱 DML——ACM2-12,见 §6)。 - wire 契约冻结:消息结构唯一事实源为 `docs/legacy/SIS_AODB_RMS-V0.1.md` + `docs/legacy/unisysaodbsis.xsd`; - HTTP 端点路径与响应语义沿用现役(如 `POST /cminmsgs/send` 返回记录 ID)。 + HTTP 端点路径与响应语义沿用现役(如 compat 写路径 `POST /cminmsgs/send` 返回记录 ID)。 - **影子对拍口径(ACM2-12)**:共享库是单信箱无法"同入口双收",影子输入改以共享库只读 水位/回放 + 自有 PG 独立 schema 比对(详见 §8 与 ACM2-12 Checks ⑤)。 +### 1.1 上下游边界(中间件职责) + +| 方向 | 角色 | 本系统做什么 | 本系统**不**做什么 | +|---|---|---|---| +| 入站(主路径) | CIIMS / AODB 等上游 | JDBC 轮询共享 MySQL `CMINMSGS` 发现新信 → 自有 PG 入队 → 解析处理 | 不生成原始业务报文;不替代 CIIMS 落信 | +| 入站(compat) | 手工工具 / 对拍 | HTTP `POST /cminmsgs/send` 写信箱 + PG 入队 | 非生产主拓扑 | +| 处理 | 本系统 | 维护 Redis 航班动态权威态;Handler 纯函数决策 | 不持有 AODB 主数据编辑权 | +| 出站 | 下游消费者 | Kafka(msg/schd)、共享 MySQL `COUTMSGS`、查询 HTTP | 不直接推送至前端(经 Kafka 等中转) | + +与 SIS / legacy 一致:上游经 CIIMS 等**外部系统**写入 `CMINMSGS` 后,本系统以 **1s JDBC 轮询** +(`DATE_PROCESSED IS NULL`)采集并处理——与 legacy `MsgExchangeRunner` 同口径。 + ## 2. 技术栈 | 层 | 选型 | 说明 | |---|---|---| | 语言/运行时 | Kotlin 2.3 + JDK 25 | JDK 21 不可行(Micronaut 5.1 系要求 JVM 25+,ACMA-9 实测) | -| 框架 | Micronaut 5.1.3 | 编译期 DI:KSP(`kotlin-ksp` + `micronaut-inject-kotlin`)生成 `*$Definition` | +| 框架 | Micronaut platform BOM **5.1.3**(core 系实际解析 **5.1.13**,classpath 混用;版本重钉属 U02/U05) | 编译期 DI:KSP(`kotlin-ksp` + `micronaut-inject-kotlin` **5.1.3**)生成 `*$Definition` | | 持久化 | **自有 PostgreSQL**(全部内部状态)+ 共享 MySQL 信箱 | 自有库:消息管道 PROC_STATE/MSG_EVENT + PUMP_JOB/REQ_TRACK + 21 类 REF_MASTER(迁移 `db/migration`);共享库仅 CMINMSGS/COUTMSGS DML。Micronaut Data JDBC 实装属 U05(ACM2-12) | | 权威存储 | Redis(航班动态 flightInfo + 快照 gen) | 仅主泵线程写(I5);Lua 原子覆盖/版本推进(gen 协议重设计属 U09) | | 投递 | Kafka(acks=all + 幂等) | outbox 模式,经 MSG_EVENT 表中转 | @@ -41,12 +57,18 @@ ## 3. 总体拓扑 ``` + CIIMS/上游 ──外部写(他人系统)──▶ 共享 MySQL CMINMSGS(信箱) + │ + │ JDBC 轮询/重扫(① 发现 DATE_PROCESSED IS NULL 新信) + ▼ ┌──────────────────────────────────────────────────┐ │ msgexchange-nextgen │ │ (单实例 · 单写者) │ - AODB/上游 ──HTTP──▶│ ingress │ - │ InboxService ──事务1──▶ CMINMSGS(原文) │ - │ └▶ PROC_STATE(PENDING) │ + │ ingress(InboxPoller + 补偿重扫,U05) │ + │ └──② 自有 PG PROC_STATE(PENDING) │ + │ (跨库非同事务;② 失败→① 重扫补建) │ + │ (compat)POST /cminmsgs/send ──▶ 信箱 insert │ + │ + PG 入队(现役 HTTP 写路径,U16 对拍) │ │ │ │ processing(msgx-pump 线程,严格 FIFO 队头) │ │ Pump ──tick──▶ MessageProcessor │ @@ -54,16 +76,19 @@ │ │ │ identity 绑定(I3) │ │ │ │ Handler.decide(纯函数) │ │ │─Schd DNLD──▶ SnapshotFlow(流程4) │ - │ │─PUMP_JOB───▶ JobExecutor(同队列,决策1) │ + │ │─PUMP_JOB───▶ JobExecutor(作业窗口,决策1) │ │ │ │ │ ├────Redis Lua──▶ Redis flightInfo(A权威) │ - │ └──事务2──▶ MSG_EVENT(outbox)+ 回填+SUCCEEDED│ + │ ├──自有 PG 事务2──▶ MSG_EVENT + SUCCEEDED │ + │ └──共享 MySQL 回填 DATE_PROCESSED(外部副作用)│ │ │ │ delivery(msgx-dispatcher 线程,每 target FIFO) │ │ Dispatcher ──逐条──▶ Kafka(msg) │ │ └─flushSchd 聚合─▶ Kafka(schd) │ │ (阶段 B 追加:ES flight_hts → Redis 投影删除) │ └──────────────────────────────────────────────────┘ + 共享 MySQL(信箱)◀──① 轮询发现 / 回填──▶ 自有 PostgreSQL ◀──② 管道状态 + Redis(动态+gen)◀── Lua 写 ── processing │ │ ▼ ▼ 下游 Kafka topic Eureka / logstash @@ -71,6 +96,12 @@ 要点: +- **收报主路径(与现役/SIS 一致)**:上游经 CIIMS 等**外部系统**写入共享 MySQL `CMINMSGS` + (本系统不建表);`ingress` 经 **JDBC 轮询**(`DATE_PROCESSED IS NULL`,1s 节律,与 + legacy `MsgExchangeRunner` 同口径)发现新信 → 自有 PG 建 `PROC_STATE(PENDING)`。 + PG 入队失败时以共享库水位**重扫补建**(U05)。`POST /cminmsgs/send` 为现役 HTTP + **写**路径(手工/对拍),非上游报文到达的主拓扑。 + - **两条专用 daemon 单线程**(`msgx-pump` / `msgx-dispatcher`)由 `PipelineLifecycle` 在 `ServerStartupEvent` 后拉起,不占用 Netty event loop;停机 `requestStop` + interrupt + join(U07)。仅当 `msgx.pipeline.autostart=true` 时装配——生产默认关, @@ -78,19 +109,22 @@ - **单写者约束(I5)**:阶段 A 全部 Redis 写集中在主泵线程;实例数必须为 1 (运行期租约/选主保护属 U26,尚未实装,当前靠部署拓扑约束)。 - **统一 FIFO(决策 1)**:定时作业(cron → PUMP_JOB 入队)与消息同队列, - 作业产物不绕过队头顺序;job 与队头的先后目前为入队时间近似, - 统一序号列属 U15(未实装,见 design.md §9 缺口清单)。 - **ACM2-12 修订**:PUMP_JOB 在自有 PG,作业不插队、仅在消息队头空闲/退避窗口由主泵执行。 + 作业产物不绕过队头顺序。**当前实现**:`jobBefore` 为 **head-state 近似**(队头为空或 + FAILED 时作业先行),非入队时间排序,与「统一 FIFO」注释存在已知偏离(U15 未实装, + 见 design.md §9)。 + **ACM2-12 修订**:PUMP_JOB 在自有 PG,作业不插队、仅在消息队头空闲/退避窗口由主泵执行 + (作业窗口语义定稿属 ACM2-12 Checks ④)。 - **存储边界(ACM2-12)**:本框图内事务库为**自有 PostgreSQL**(PROC_STATE/MSG_EVENT/ - PUMP_JOB/REQ_TRACK/21 类);CMINMSGS/COUTMSGS 在**共享 MySQL 信箱**(外部副作用,非本系统 - 表);Redis 除 flightInfo 还承载快照 gen。图中"事务 1/2"与 ACM2-3 的"同库事务"语义不同, - 见 §6:本地事务只在自有 PG,信箱交互为外部写+最终一致。 + PUMP_JOB/REQ_TRACK/21 类);CMINMSGS/COUTMSGS 在**共享 MySQL 信箱**(上游外部写、 + 本系统 JDBC 轮询读 + 处理回填写);Redis 除 flightInfo 还承载快照 gen。图中跨库步骤 + (① JDBC 发现 / ② PG 入队 / 外部回填)与 ACM2-3 原「同库事务锚」语义不同,见 §6: + 本地事务只在自有 PG,信箱交互为外部读/写 + 最终一致。 ## 4. 模块职责 | 包 | 职责 | 对应 ACMA-8 | 主要类 | |---|---|---|---| -| `ingress/` | 收报:共享信箱写原文 + 自有 PG 建 PENDING(跨库外部副作用,ACM2-12);不解析报文 | 流程 1,I3 | `InboxController` `InboxService` | +| `ingress/` | 收报:JDBC 轮询共享信箱发现新信 → 自有 PG 建 PENDING(+ 补偿重扫;HTTP 写路径 compat);不解析报文 | 流程 1,I3 | `InboxPoller`(U05)`InboxController` `InboxService` | | `processing/` | 主泵:FIFO 领取、解码、identity 绑定、纯函数决策、自有 PG 事务2 | 流程 2/4,I1/I2/I5 | `Pump` `MessageProcessor` `SnapshotFlow` `Identity` `Handler(Registry)` | | `delivery/` | 投递:每 target 严格 FIFO、schd 聚合 | 流程 3 | `Dispatcher` `SchdAggregation` | | `jobs/` | 泵作业:清场/归档/投影重建(PUMP_JOB 自有 PG,作业窗口执行) | 流程 4/5/7,I4 | `JobExecutor` `HistorySweepJob` `ArchiveJob` `ProjectionRebuildJob` | @@ -119,9 +153,10 @@ 状态/重试/毒丸)+ `MSG_EVENT`(发消息侧:outbox)+ `PUMP_JOB`(作业调度)+ `REQ_TRACK` (15 类请求)+ `REF_MASTER`(21 类静态,SOURCE 审计)。 - **共享 MySQL(cdairport,他人系统库)——本系统不建任何表/schema,仅信箱 DML**: - 收报写 `CMINMSGS`(含处理完成回填 DATE_PROCESSED/STATUS)、出站写 `COUTMSGS` - (他人系统读取发送)。与信箱的交互是**外部副作用**,非本系统事务的一部分: - 收报 = 信箱落信成功(返回 CMINMSGS_ID)→ 自有 PG 建 PENDING 入队;PG 建行失败以共享库 + 上游外部写 `CMINMSGS`;本系统 JDBC 轮询读 + 处理完成回填 `DATE_PROCESSED/STATUS`; + 出站写 `COUTMSGS`(他人系统读取发送)。与信箱的交互是**外部副作用**,非本系统事务的一部分: + 收报主路径=JDBC 轮询发现新信 → 自有 PG 建 PENDING 入队;compat HTTP 写= + 信箱 insertRaw 成功(返回 CMINMSGS_ID)→ PG 入队;PG 建行失败以共享库 `DATE_PROCESSED IS NULL` 重扫补建;回填 = 处理成功后异步/补偿,最终一致(ACM2-12 影响面)。 - **Redis**:航班动态 flightInfo(阶段 A 权威,I5)+ 快照 **SCHD_GEN**(Lua 内原子 「覆盖+按代差删+版本推进」,协议重设计属 U09)。 @@ -138,7 +173,7 @@ | B(`msgx.phase=B`) | —(缓做,ACM2-12) | ES 历史投影 | 不落表、不启用;Redis 永续动态权威,阶段 B 价值重评估后再定 | **就绪度(2026-09-07 复核口径)**:可编译、37 测试全绿、**dev stub 进程级冒烟实测可端到端** -(`./gradlew run` 无外部依赖启动 → 收报 200 → `/health` UP,修复记录见 README「进程级 dev 冒烟」); +(`./gradlew run` 无外部依赖启动 → compat HTTP 写路径返回 200 → `/health` UP,修复记录见 README「进程级 dev 冒烟」); 生产默认配置**不可对外服务**——`autostart=false` 且生产(stubs=false)下仓储无实装、DI 装配 即失败。生产就绪前置:U05(数据层+事务)、U07 fail-fast 定案、U09(快照恢复协议)、 U13(投递毒丸补全)、U15(统一序号)。逐项状态见 ACM2-10「定稿实施计划」。 @@ -156,13 +191,13 @@ U13(投递毒丸补全)、U15(统一序号)。逐项状态见 ACM2-10「 - **管理端点**:Micronaut 5.1 下 `/env` 默认**禁用**、`/beans` 默认 enabled+sensitive;dev/影子经 顶层 `endpoints.*`(**非** `micronaut.endpoints.*`——实测前缀错误时不生效)放开 `/env`、`/beans` 与 health 明细。工程未引入 micronaut-security,sensitive 的实际拦截行为待 U28 定案。 -- **同名单风险**:影子与生产同名同路径会互相收报——切流前必须核对服务名三隔离。 +- **同名单风险**:影子与生产同名同路径时,compat HTTP 写可能误写生产信箱——切流前必须核对服务名三隔离。 ## 9. 可观测性 - **日志**:logstash TCP JSON 通道(Async + neverBlock 降级,logstash 不可达不阻塞业务线程); - 结构化生命周期日志(收报/SUCCEEDED/SKIPPED/FAILED/DEAD/毒丸/flush 批次);MDC `traceId` - (当前 = cminmsgsId/eventId,处理片段;贯穿收报→投递属 U12 遗留)。 + 结构化生命周期日志(入队/PENDING/SUCCEEDED/SKIPPED/FAILED/DEAD/毒丸/flush 批次);MDC `traceId` + (当前 = cminmsgsId/eventId,处理片段;贯穿入队→投递属 U12 遗留)。 - **健康**:`/health` 聚合 `redis-flight-store` / `kafka-delivery` 自定义指示器——真实 ping 判定(false/异常→DOWN,缺 bean→DOWN),非仅 bean 存在。 - **指标缺口**:micrometer 队列深度/投递延迟 gauge 未引入(版本对齐待 U05 批次); diff --git a/docs/design.md b/docs/design.md index d52b79d..e2bfb90 100644 --- a/docs/design.md +++ b/docs/design.md @@ -1,9 +1,33 @@ # msgexchange-v2 设计文档 +> **系统角色**:机场 OMMS **上游报文处理中间件**——消费 CIIMS/AODB 等上游经共享 MySQL 信箱 +> (`CMINMSGS`)投递的 XML 报文,解析处理后维护 Redis 动态并向 Kafka / 出站信箱投递; +> **非**报文源系统。生产主路径 = **JDBC 轮询**发现新信;HTTP `POST /cminmsgs/send` = compat 写路径。 > 本文对应仓库当前实现,给出模块级设计语义与依据;架构总览见 > [architecture.md](architecture.md),权威架构为 Plane **ACM2-3**,实施计划与逐项验收为 > **ACM2-10(U01–U30)**。文中标注「TODO/未实装」的条目均为已知开放项,不属文档遗漏。 +## 0. 系统边界速览 + +``` +上游(CIIMS/AODB…) ──外部写──▶ 共享 MySQL CMINMSGS(DATE_PROCESSED IS NULL) + │ JDBC 轮询/重扫(InboxPoller,U05) + ▼ + 自有 PG PROC_STATE(PENDING) ──▶ 主泵 FIFO 处理 + │ + ┌───────────────┼───────────────┐ + ▼ ▼ ▼ + Redis 动态 Kafka msg/schd COUTMSGS 出站 + (阶段 A 权威) (下游订阅) (他人读取发送) + +(compat)POST /cminmsgs/send ──▶ insertRaw + PG 入队(手工/对拍,非主拓扑) +``` + +- **中间件定位**:本系统位于 CIIMS 与下游消费者之间,负责**采集 → 解析 → 决策 → 投递**; + 报文原文由上游写入共享信箱,本系统只读(主路径)或 compat 写(辅助)。 +- **与 legacy 对齐**:legacy `MsgExchangeRunner` 同样以 1s 轮询 `CMINMSGS` 为处理入口; + legacy HTTP 收报接口在 nextgen 中保留为 compat,不改变生产主拓扑。 + ## 1. 领域模型 ### 1.1 状态机与错误分类 @@ -65,8 +89,8 @@ ErrorClass(两侧共用): **存储边界(ACM2-12 定案)**: - **自有 PostgreSQL** = 上表全部(消息管道 + 调度 + 请求 + 21 类)。本地事务只在此库: 处理侧「MSG_EVENT 插入 + PROC_STATE→SUCCEEDED」同事务;其余跨存储一律外部副作用。 -- **共享 MySQL(cdairport,他人系统)仅信箱 DML、不建表**:收报写 CMINMSGS、出站写 - COUTMSGS(他人读取发送)。见 §3.1/§3.2 的事务模型。 +- **共享 MySQL(cdairport,他人系统)仅信箱 DML、不建表**:上游外部写 CMINMSGS; + 本系统 JDBC 轮询读 + 处理回填;出站写 COUTMSGS(他人读取发送)。见 §3.1/§3.2 的事务模型。 - **Redis**:航班动态 flightInfo + 快照 **SCHD_GEN(gen)**——Lua 内原子「覆盖+按代差删+ 版本推进」;重放幂等由 Lua 承接(协议重设计属 U09),`RefDataRepository` 为目标实现的 过渡占位接口。 @@ -77,14 +101,20 @@ ErrorClass(两侧共用): ## 3. 核心流程设计 -### 3.1 流程 1:收报(`InboxService.accept`) +### 3.1 流程 1:收报(JDBC 轮询 + HTTP compat) -收报 = 共享信箱外部写 + 自有 PG 入队(ACM2-12,跨库,非同一事务): -1. `insertRaw`:共享 MySQL CMINMSGS 落信成功(返回其主键 CMINMSGS_ID)——「已持久化」 - 语义即刻成立,响应可返回(与现役逐字对拍后固化,U16); -2. `procState.insert(id)`:自有 PG 建 PENDING 行入队;本步失败 → 补偿(以共享库 - `DATE_PROCESSED IS NULL` 重扫补建,U05 批次实现); -3. 不解析报文、接收层无唯一约束(I3)。`wakePump()` 为 TODO 空操作——泵 1s 轮询兜底。 +**主路径(生产/SIS 口径)**:上游经 CIIMS 等外部系统写入共享 MySQL `CMINMSGS` +(`DATE_PROCESSED IS NULL`);本系统 `ingress` 经 **JDBC 轮询**发现新信(与 legacy +`MsgExchangeRunner.getNewMsgsAfterId` 同语义,1s 节律),自有 PG 入队: + +1. `pollNew()`(U05,`InboxPoller`):JDBC 查共享库 `CMINMSGS_ID > watermark 且 + DATE_PROCESSED IS NULL`(及/或 PG 无对应 PROC_STATE 的补偿重扫); +2. `procState.insert(id)`:自有 PG 建 PENDING 行入队;本步失败 → 下轮重扫补建; +3. 不解析报文、接收层无唯一约束(I3);`wakePump()` 为 TODO 空操作——主泵 1s 轮询兜底。 + +**compat 路径(现役 HTTP 写)**:`InboxService.accept`(`POST /cminmsgs/send`)= +共享信箱 `insertRaw` + 自有 PG 入队(跨库,非同一事务);「已持久化」响应语义与现役 +对拍(U16)。用于手工注入/影子对拍,**非**上游报文到达的主拓扑。 ### 3.2 流程 2:主泵 tick(`Pump.tick`) @@ -92,8 +122,9 @@ ErrorClass(两侧共用): 1. `headQueued()` 取作业、`headUnfinished()` 取最小未完成 CMINMSGS_ID(**含 FAILED**, 消息间严格保序:队头退避未到期即 sleep 至到期点,后方消息永不越队)。 -2. `jobBefore(job, head)`:head 为空或队头 FAILED 时作业先行——**当前为入队时间近似, - 与「统一 FIFO」注释存在已知偏离(U15 未实装)**;统一序号列定案后消除。 +2. `jobBefore(job, head)`:head 为空或队头 FAILED 时作业先行——**当前为 head-state 近似** + (非入队时间排序),与「统一 FIFO」注释存在已知偏离(U15 未实装);ACM2-12 口径为 + 作业窗口执行(队头空闲/退避窗口),不追求与消息统一全序(Checks ④)。 3. 队头 FAILED 且退避未到期:`poisoned()` 判定(attempts≥maxAttempts 或滞留超 head-deadline 10m)→ DEAD(EXHAUSTED) 毒丸升级(Pump 侧;投递侧同语义属 U13,未实装); **实现注**:headDeadline 判据以 `updatedAt` 为锚,而 updatedAt 与 nextAttemptAt 同一次 FAILED @@ -107,8 +138,9 @@ ErrorClass(两侧共用): 其余 → `Handler.decide(redis.hgetAllFlightInfo(), msg)` → 阶段 A:Redis 先写(I2 happens-before,TODO redisApply)→ 自有 PG 事务 2: MSG_EVENT 插入 + PROC_STATE→SUCCEEDED(同库原子,@Transactional); - CMINMSGS 回填(DATE_PROCESSED/STATUS)为共享信箱**外部回填**:PG 提交后异步/补偿执行, - 失败重试+告警(最终一致,ACM2-12;回滚接管语义以回填成功为界,需 Runbook 演练确认)。 + CMINMSGS 回填(DATE_PROCESSED/STATUS)为共享信箱**外部回填**:目标态为 PG 提交后 + 异步/补偿执行,失败重试+告警(最终一致,ACM2-12);**当前实现为同步内联占位** + (`insertAll → backfillOnSuccess → SUCCEEDED`,无 `@Transactional`,U05 定案并改)。 5. 异常边界(U08):`processOne` 内 try/catch → `ProcFailure.fail(INFRA)`(attempts+1、 退避、达上限 DEAD);`InterruptedException` 恢复中断位后**上抛**;loop 仅 catch `Exception` 作最后防线,`Error` 任其终止进程(异常必可见)。 @@ -168,7 +200,7 @@ staging(流式解析+整包校验,TODO 阶段2;未实装→FAILED(UNSUPPOR | 不变量 | 语义 | 落点 | 状态 | |---|---|---|---| -| I1 | 单写者严格 FIFO + HOL 阻塞 + 毒丸升级 | `headUnfinished`/`headUnsent` 队头语义、`poisoned()` | 实装(attempts 毒丸生效;head-deadline 判据不可达待修,见 §3.2 注;job 相对队头的全序属 U15) | +| I1 | 单写者严格 FIFO + HOL 阻塞 + 毒丸升级 | `headUnfinished`/`headUnsent` 队头语义、`poisoned()` | 实装(attempts 毒丸生效;head-deadline 判据不可达待修,见 §3.2 注;job 为 head-state 近似 / 作业窗口属 U15) | | I2 | Redis 先写、后于事件创建(happens-before) | `processOne` 阶段 A 分支 | TODO redisApply(流程占位已留) | | I3 | identity 首绑幂等;接收层无唯一约束;SUCCEEDED 回填 | `Identity`/`tryBindIdentity`/`backfillOnSuccess` | 实装 | | I4 | 清场仅删 ES 成功集;按代差删 | `HistorySweepJob`/SNAPSHOT_REPLACE delFields | 门禁实装,ES 接线 TODO | @@ -220,11 +252,11 @@ ACM2-12)、`flyway.datasources.default.*`、`mailbox.shared-mysql.*`(共享 | 项 | 缺口 | 计划 | |---|---|---| -| U05 | 自有 PG 数据层无实装(PROC_STATE/MSG_EVENT/PUMP_JOB/REQ_TRACK/REF_MASTER + 信箱适配层 CminmsgMailbox/OutboxMailbox);`@Transactional`/allopen 未引入 → 生产 DI 装配失败 | U05 批次(Micronaut Data JDBC on PG + allopen + 信箱外部副作用与补偿回归) | +| U05 | 自有 PG 数据层无实装 + **InboxPoller(JDBC 轮询/重扫)** 未实装(当前仅 compat HTTP `InboxService.accept`);信箱适配层 CminmsgMailbox/OutboxMailbox;`@Transactional`/allopen 未引入 → 生产 DI 装配失败;`application-dev.yml` 仍残留 MySQL datasource URL 占位(stub 下 enabled=false 不建连,U05 清理) | U05 批次(Micronaut Data JDBC on PG + InboxPoller + allopen + 信箱外部副作用与补偿回归) | | U07/U26 | `autostart` 默认关=有意门禁,但生产无 fail-fast;双实例无运行期防护 | fail-fast 定案 + 租约/DB 锁拒启 | | U09 | gen→Redis 协议未重设计(Lua 内原子版本推进;崩溃窗口=「Lua 完成/PG SUCCEEDED 未写」) | Redis 内版本 CAS + 恢复协议 + 测试(ACM2-12) | | U13 | 投递侧无 createdAt/headDeadline 超时升级、无 DEAD 告警出口;处理侧 head-deadline 判据亦不可达(见 §3.2 注) | WP2 | -| U15 | job 与队头消息无统一全序(入队时间近似)——ACM2-12 后改口径:作业窗口执行,不追求与消息全序 | 作业窗口语义定稿(ACM2-12 Checks ④) | +| U15 | job 与队头消息无统一全序(**head-state 近似**,非入队时间)——ACM2-12 口径:作业窗口执行,不追求与消息全序 | 作业窗口语义定稿(ACM2-12 Checks ④) | | U16 | `/cminmsgs/send` 无 @Consumes/字符集(实测 text/plain 415)、无错误路径契约(@ControllerAdvice) | WP2(legacy 逐字对拍固化) | | U17 | Eureka 注册名仍取 `micronaut.application.name`(=msgexchange-nextgen);`msgx.service-name` 无运行时消费方 → 影子/切流前注册名与文档契约脱节 | `micronaut.application.name=${msgx.service-name}`(application.yml) | | U18 | DDL 缺口(随 PG 化收窄:REQ_TRACK.COUTMSGS_ID 已 BIGINT;时间列口径 U05 定) | U05 批次 | @@ -232,5 +264,5 @@ ACM2-12)、`flyway.datasources.default.*`、`mailbox.shared-mysql.*`(共享 | U22–U24 | 载荷类型收敛、eventSeq 未接线、每报文全量读语义定案 | WP3 | | U25/U28/U30 | 一致性哨兵实装、README 安全节/入口、索引与杂项 | WP3/4 | | U27 | 专有材料(SIS md 703KB / XSD 版权头)治理决策 | WP4(ACL 核验先行) | -| ACM2-11/12 | 存储边界(自有 PG + 共享信箱 + Redis 动态/gen + 阶段 B 缓做):迁移 SQL/配置/接口注释已按定案调整(V1.0.0 PG);信箱适配层、gen Lua、作业窗口语义、影子重设计未实装 | ACM2-12 Checks ①–⑥ | -| ACM2-11 | 21 类静态主数据独立 PG 参考库(datasources.reference):Repository 接口已拆(RefDataRepository=gen-only / StaticRefRepository),表结构/迁移/SOURCE 审计/ReferenceService 同步与应答接线未实装 | 阶段 6(U05 批次后);生产 Oracle 仅可能性,触发条件见 ACM2-11 | +| ACM2-12 | 存储边界(自有 PG + 共享信箱 + Redis 动态/gen + 阶段 B 缓做):迁移 SQL/配置/接口注释已按定案调整(V1.0.0 PG);信箱适配层、gen Lua、作业窗口语义、影子重设计未实装 | ACM2-12 Checks ①–⑥ | +| ACM2-11 | **决策史(已被 ACM2-12 吸收)**:曾讨论 21 类静态独立 PG 参考库;定案为并入自有 PG `REF_MASTER`(见 ACM2-12),勿再按 `datasources.reference` 第二库规划 | 仅作决策脉络参考 | diff --git a/docs/legacy/msgexchange-api-legacy-user-stories.md b/docs/legacy/msgexchange-api-legacy-user-stories.md index 5497ee8..9b4cd0e 100644 --- a/docs/legacy/msgexchange-api-legacy-user-stories.md +++ b/docs/legacy/msgexchange-api-legacy-user-stories.md @@ -2,6 +2,11 @@ ## 0. 文档说明 +> **nextgen 定位注(msgexchange-v2)**:本文件描述 **legacy** 行为基线。nextgen 为**上游报文 +> 处理中间件**——生产主路径 = JDBC 轮询共享 MySQL `CMINMSGS`(上游经 CIIMS 等外部落信); +> legacy §3.1 HTTP 收报与 §3.2 轮询在现役并存,nextgen 将轮询提升为主拓扑、HTTP 降为 compat +> (见仓库 `docs/architecture.md`)。 + - 本文档记录 msgexchange-api(现役旧项目)的完整业务逻辑,并以用户故事形式组织,作为新系统(Micronaut + Kotlin)迁移设计及一致性对拍的行为基线。 - 内容依据仓库源码与资源配置编写:`src/main/java/com/gzzn/omms/msgexchange-api/**`、`src/main/resources/*.yml`;分支 `feature/add_dev_containers_supports`(HEAD `eb72a668`)。 - 报文结构依据 `SIS_AODB_RMS-V0.1.md` 与 `unisysaodbsis.xsd`(同目录 docs/legacy/)。 diff --git a/settings.gradle.kts b/settings.gradle.kts index b06182b..9ae3a5d 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -7,5 +7,5 @@ pluginManagement { } } -// 新一代消息交换服务:ACMA-8 v4 架构(单写者严格 FIFO 管道 + 两阶段权威)。 +// 机场 OMMS 上游报文处理中间件:JDBC 轮询共享 CMINMSGS 主路径 + FIFO 管道(ACMA-8 v4)。 // 独立仓库,与 legacy msgexchange-api 并存至切流完成。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt index cdb32fd..f0cb513 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt @@ -12,7 +12,8 @@ import java.time.Instant * ACM2-12 仓储接口(接口驱动,主泵/调度循环可单测;Micronaut Data JDBC 实装属 U05 批次)。 * 存储边界:自有 PostgreSQL(datasources.default)= 本文件除 CminmsgInboxRepository 外 * 的全部接口(消息管道 PROC_STATE/MSG_EVENT、PUMP_JOB、REQ_TRACK、21 类 REF_MASTER); - * 共享 MySQL 信箱(CMINMSGS 收 / COUTMSGS 出)经信箱封装访问,仅 DML、不建表; + * 共享 MySQL 信箱(CMINMSGS / COUTMSGS)经信箱封装访问,仅 DML、不建表; + * 主路径=上游外部写 CMINMSGS → 本系统 JDBC 轮询读;compat=insertRaw HTTP 写; * Redis = 航班动态 + 快照 gen(RefDataRepository 目标实现);FLIGHT_STATE 缓做。 */ interface ProcStateRepository { @@ -148,15 +149,20 @@ interface FlightStateRepository { } /** - * 收报信箱(共享 MySQL CMINMSGS,仅 DML——ACM2-12:库属他人系统,本系统不建表)。 - * 事务模型:insertRaw = 外部副作用「信箱落信」(成功即返回其主键 CMINMSGS_ID)→ - * 随后自有 PG 建 PROC_STATE(PENDING) 入队;PG 建行失败以共享库 - * DATE_PROCESSED IS NULL 重扫补建。backfillOnSuccess 为处理成功后的外部回填 - * (DATE_PROCESSED/STATUS,最终一致、失败重试+告警)。 + * 共享信箱 CMINMSGS 访问(ACM2-12:库属他人系统,本系统不建表)。 + * + * **主路径(生产)**:`pollNew` / `rawOf` — JDBC 轮询/读取上游外部写入的报文(U05 InboxPoller)。 + * **compat 路径**:`insertRaw` — HTTP `POST /cminmsgs/send` 辅助写(手工/对拍,非主拓扑)。 + * + * 入队模型:发现或 compat 写得到 CMINMSGS_ID → 自有 PG 建 PROC_STATE(PENDING); + * PG 建行失败以共享库 DATE_PROCESSED IS NULL 重扫补建。 + * `backfillOnSuccess` = 处理成功后的外部回填(DATE_PROCESSED/STATUS,最终一致)。 */ interface CminmsgInboxRepository { + /** compat HTTP 写路径:向共享信箱插入原文(U16 契约对拍)。 */ fun insertRaw(rawXml: String): Long + /** 主路径/处理:按 CMINMSGS_ID 读取上游已落信的原文 XML。 */ fun rawOf(cminmsgsId: Long): String? fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) 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 index 89da81f..0123713 100644 --- 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 @@ -21,7 +21,7 @@ import java.util.concurrent.atomic.AtomicLong /** * U07(U01 遗留 stub):内存仓储装配——仅当 `msgx.stubs=true`(dev/影子冒烟)时生效 - * (@Requires),让「启动 + 收报 + 主泵/投递循环 + /beans 端到端」在无 MySQL/Redis/Kafka 时可跑。 + * (@Requires),让「启动 + compat HTTP 写 + 主泵/投递循环 + /beans 端到端」在无 MySQL/Redis/Kafka 时可跑。 * 与真实 Micronaut Data 实装按模块替换;切换点条件显式,不误入生产。 */ @Requires(property = "msgx.stubs", value = "true") diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt index 7de4814..fa4d375 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxController.kt @@ -8,7 +8,8 @@ import io.micronaut.http.annotation.Post import io.micronaut.http.annotation.Produces /** - * ACMA-8 契约冻结:端点路径与响应语义不变。现役 Controller 返回记录 ID(原文落库后)。 + * ACMA-8 契约冻结:compat HTTP 写路径 `POST /cminmsgs/send`(手工/对拍)。 + * 生产主路径 = JDBC 轮询共享 CMINMSGS(InboxPoller,U05),非本 Controller。 */ @Controller class InboxController(private val inbox: InboxService) { diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt index 0289a35..b0edc56 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt @@ -5,21 +5,20 @@ import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository import jakarta.inject.Singleton import java.time.Instant -/** ACMA-8 流程 1:入站接收。响应语义 = “已持久化”(与现役一致);不解析报文(I3)。 */ +/** ACMA-8 流程 1 · compat 写路径:HTTP 落信 + PG 入队(非生产主拓扑;主路径=InboxPoller JDBC 轮询,U05)。 */ @Singleton class InboxService( private val inbox: CminmsgInboxRepository, private val procState: ProcStateRepository, - // TODO(U05 批次): 事务模型(ACM2-12)——insertRaw 为共享信箱外部写(成功即返回 ID), - // 随后自有 PG 建 PROC_STATE(PENDING);两写跨库,PG 建行失败以共享库 - // DATE_PROCESSED IS NULL 重扫补建;pump 唤醒仅加速,崩溃后主泵 1s 轮询兜底。 + // TODO(U05): 主路径 InboxPoller — JDBC 轮询 DATE_PROCESSED IS NULL → procState.insert; + // compat 路径 insertRaw + PG 入队跨库;PG 失败以共享库重扫补建;wakePump 仅加速。 ) { private val log = org.slf4j.LoggerFactory.getLogger(InboxService::class.java) data class Receipt(val cminmsgsId: Long, val receivedAt: Instant) fun accept(rawXml: String): Receipt { - val id = inbox.insertRaw(rawXml) // 信箱外部写(ACM2-12,与 PG 跨库) + val id = inbox.insertRaw(rawXml) // compat:共享信箱外部写(ACM2-12,与 PG 跨库) procState.insert(id) // 自有 PG 入队;接收层无唯一约束(I3) wakePump() log.info("accepted cminmsgsId={}", id) diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index 14eb971..475d1e8 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -1,6 +1,7 @@ # 开发/影子对拍环境 profile(U03/N32 + U07): # - /env、/beans 放开以便“改值—回读”与装配核验(生产 profile 不加载本文件) -# - msgx.stubs=true:无 MySQL/Redis/Kafka 时以内存 stub 仓储跑通「启动 + 收报 + 主泵/投递循环」 +# - msgx.stubs=true:无 MySQL/Redis/Kafka 时以内存 stub 跑通「启动 + compat HTTP 写 + +# 主泵/投递循环」(生产主路径 JDBC 轮询 InboxPoller 属 U05,stub 未覆盖) # - msgx.pipeline.autostart=true:启动即拉起 Pump/Dispatcher 专用线程 # 启动:MICRONAUT_ENVIRONMENTS=dev ./gradlew run msgx: diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 0417f73..425db47 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,5 +1,7 @@ # ===================================================================== # msgexchange-nextgen 配置(ACM2-12 口径) +# 系统角色:机场 OMMS 上游报文处理中间件(JDBC 轮询共享 CMINMSGS 为主路径; +# compat HTTP POST /cminmsgs/send 为辅助写路径)。 # 口令/端点全部环境变量外置——零入库。 # 存储边界:自有 PostgreSQL(datasources.default,全部内部状态)+ # 共享 MySQL 信箱(mailbox.cminmsgs / mailbox.coutmsgs,仅 DML,不建表)+ @@ -51,8 +53,9 @@ flyway: enabled: false locations: classpath:db/migration -# 共享 MySQL 信箱(他人系统库,仅 DML:收报写 CMINMSGS、出站写 COUTMSGS;不建表/schema, -# ACM2-12)。驱动/依赖与信箱适配层(CminmsgMailbox/OutboxMailbox)随 U05 批次引入。 +# 共享 MySQL 信箱(他人系统库,仅 DML:上游外部写 CMINMSGS;本系统 JDBC 轮询读 + +# 处理回填;出站写 COUTMSGS;不建表/schema,ACM2-12)。驱动/依赖与信箱适配层 +# (CminmsgMailbox/OutboxMailbox)随 U05 批次引入。 mailbox: shared-mysql: url: ${MSGX_MAILBOX_URL} diff --git a/src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql b/src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql index ac36411..530f2a8 100644 --- a/src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql +++ b/src/main/resources/db/migration/V1.0.0__own_pg_pipeline.sql @@ -5,8 +5,9 @@ -- 六表脚本已删除,定义见 git 历史): -- · 自有 PG:全部内部状态 —— 消息管道(PROC_STATE / MSG_EVENT)+ 调度与请求 -- (PUMP_JOB / REQ_TRACK)+ 21 类静态主数据(REF_MASTER)。本地事务只在此库。 --- · 共享 MySQL(cdairport,他人系统库):仅信箱 DML —— 收报写 CMINMSGS、 --- 出站写 COUTMSGS(他人读取发送);本系统不在此建表/schema(迁移不覆盖)。 +-- · 共享 MySQL(cdairport,他人系统库):仅信箱 DML —— 上游外部写 CMINMSGS; +-- 本系统 JDBC 轮询读 + 处理回填;出站写 COUTMSGS(他人读取发送); +-- 本系统不在此建表/schema(迁移不覆盖)。 -- · Redis:航班动态 flightInfo + 快照 SCHD_GEN(Lua 原子,另行脚本)。 -- · FLIGHT_STATE(阶段 B 权威):缓做,不落表(Redis 永续动态权威)。 -- 21 类静态表 REF_MASTER 为通用 rtype/rkey 形态(SOURCE 审计;gen/SCHD_GEN @@ -14,7 +15,8 @@ -- 影子对拍:本库开独立 schema/实例;共享信箱只读水位/回放(见 ACM2-12 影响面)。 -- ===================================================================== --- ① 收报处理伴生状态(取消息侧;与共享库 CMINMSGS 一一对应,CMINMSGS_ID 沿用其主键) +-- ① 处理伴生状态(取消息侧;与共享库 CMINMSGS 一一对应,CMINMSGS_ID 沿用其主键; +-- 入队由 JDBC 轮询发现新信后写入,非本系统创建 CMINMSGS 原文) CREATE TABLE PROC_STATE ( CMINMSGS_ID BIGINT PRIMARY KEY, STATE VARCHAR(16) NOT NULL, -- PENDING/FAILED/SUCCEEDED/SKIPPED/DEAD diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt index 1f33085..9bfc889 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineSmokeTest.kt @@ -20,8 +20,9 @@ import org.junit.jupiter.api.Test /** * U07 端到端(stub 装配,等价 /beans 核验):无 MySQL/Redis/Kafka 下, - * ①核心 bean 装配齐全;②收报→落库→主泵领取→解码未实装→FAILED(CODEC_ERROR)+退避; + * ①核心 bean 装配齐全;②compat HTTP 写→PG 入队→主泵领取→解码未实装→FAILED(CODEC_ERROR)+退避; * ③U11 重放把 FAILED 拉回 PENDING;④Dispatcher flushSchd 聚合发出 schd。 + * (生产主路径 JDBC 轮询 InboxPoller 属 U05,本测未覆盖。) * 后台循环关闭(autostart=false),按需手动 tick,避免测试泄漏线程。 */ @MicronautTest