shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add fact payloads to kafka envelope

Browse Source
This commit is contained in:
lingniu
2026-06-29 12:58:27 +08:00
parent 0ec29a732c
commit 9c2779f82b
3 changed files with 108 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
modules/sinks/sink-mq/pom.xml
View File

@@ -16,6 +16,10 @@
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-api</artifactId>
</dependency>
<dependency>
<groupId>com.lingniu.ingest</groupId>
<artifactId>ingest-facts</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>

39
modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto
View File

@@ -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<string, string> 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<string, string> fields = 8;
map<string, string> metadata = 9;
}
message RealtimePayload {
optional double speed_kmh = 1;
optional double total_mileage_km = 2;

65
modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/VehicleEnvelopeProtoCompatibilityTest.java Normal file
View File

@@ -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");
}
}