diff --git a/modules/sinks/sink-mq/pom.xml b/modules/sinks/sink-mq/pom.xml index 3e81b45a..03a9d565 100644 --- a/modules/sinks/sink-mq/pom.xml +++ b/modules/sinks/sink-mq/pom.xml @@ -16,6 +16,10 @@ com.lingniu.ingest ingest-api + + com.lingniu.ingest + ingest-facts + org.springframework.boot spring-boot-starter diff --git a/modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto b/modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto index 8dc92256..67dcadb8 100644 --- a/modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto +++ b/modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto @@ -35,6 +35,10 @@ message VehicleEnvelope { // 全字段内部遥测快照。新下游消费者优先读取这里,避免依赖协议字段名。 TelemetrySnapshot telemetry_snapshot = 50; + + // 新一代事实模型:RAW 帧索引与解析事实,均不携带完整 raw bytes。 + RawFrameFactPayload raw_frame_fact = 60; + DecodedFactPayload decoded_fact = 61; } message RawArchiveRef { @@ -58,6 +62,41 @@ message TelemetryField { string source_path = 6; } +enum ParseStatusProto { + PARSE_STATUS_UNSPECIFIED = 0; + PARSE_STATUS_NOT_PARSED = 1; + PARSE_STATUS_SUCCEEDED = 2; + PARSE_STATUS_FAILED = 3; +} + +message RawFrameFactPayload { + string frame_id = 1; + string vehicle_key = 2; + string vin = 3; + string phone = 4; + int32 message_id = 5; + int32 sub_type = 6; + string raw_uri = 7; + string checksum = 8; + int64 raw_size_bytes = 9; + ParseStatusProto parse_status = 10; + string parse_error = 11; + string peer = 12; + map metadata = 13; +} + +message DecodedFactPayload { + string fact_id = 1; + string frame_id = 2; + string fact_type = 3; + string vehicle_key = 4; + string vin = 5; + string phone = 6; + string raw_uri = 7; + map fields = 8; + map metadata = 9; +} + message RealtimePayload { optional double speed_kmh = 1; optional double total_mileage_km = 2; diff --git a/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/VehicleEnvelopeProtoCompatibilityTest.java b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/VehicleEnvelopeProtoCompatibilityTest.java new file mode 100644 index 00000000..c68d5a76 --- /dev/null +++ b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/VehicleEnvelopeProtoCompatibilityTest.java @@ -0,0 +1,65 @@ +package com.lingniu.ingest.sink.mq; + +import com.lingniu.ingest.sink.mq.proto.DecodedFactPayload; +import com.lingniu.ingest.sink.mq.proto.ParseStatusProto; +import com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload; +import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class VehicleEnvelopeProtoCompatibilityTest { + + @Test + void rawFrameFactPayloadRoundTripsThroughEnvelope() throws Exception { + VehicleEnvelope envelope = VehicleEnvelope.newBuilder() + .setSchemaVersion("2.0") + .setEventId("frame-1") + .setVin("VIN001") + .setSource("GB32960") + .setEventTimeMs(1782662400000L) + .setIngestTimeMs(1782662400001L) + .setRawFrameFact(RawFrameFactPayload.newBuilder() + .setFrameId("frame-1") + .setVehicleKey("VIN001") + .setMessageId(2) + .setRawUri("archive://raw/2026/06/29/GB32960/VIN001/frame-1.bin") + .setChecksum("sha256:abc") + .setRawSizeBytes(128) + .setParseStatus(ParseStatusProto.PARSE_STATUS_SUCCEEDED) + .build()) + .build(); + + VehicleEnvelope parsed = VehicleEnvelope.parseFrom(envelope.toByteArray()); + + assertThat(parsed.hasRawFrameFact()).isTrue(); + assertThat(parsed.getRawFrameFact().getFrameId()).isEqualTo("frame-1"); + assertThat(parsed.getRawFrameFact().getParseStatus()).isEqualTo(ParseStatusProto.PARSE_STATUS_SUCCEEDED); + } + + @Test + void decodedFactPayloadRoundTripsThroughEnvelope() throws Exception { + VehicleEnvelope envelope = VehicleEnvelope.newBuilder() + .setSchemaVersion("2.0") + .setEventId("fact-1") + .setVin("VIN001") + .setSource("JT808") + .setEventTimeMs(1782662400000L) + .setIngestTimeMs(1782662400001L) + .setDecodedFact(DecodedFactPayload.newBuilder() + .setFactId("fact-1") + .setFrameId("frame-1") + .setFactType("location") + .setVehicleKey("jt808:013912345678") + .setRawUri("archive://raw/2026/06/29/JT808/jt808_013912345678/frame-1.bin") + .putFields("longitude", "120.1") + .putFields("latitude", "30.1") + .build()) + .build(); + + VehicleEnvelope parsed = VehicleEnvelope.parseFrom(envelope.toByteArray()); + + assertThat(parsed.hasDecodedFact()).isTrue(); + assertThat(parsed.getDecodedFact().getFieldsMap()).containsEntry("longitude", "120.1"); + } +}