diff --git a/modules/core/ingest-core/pom.xml b/modules/core/ingest-core/pom.xml
index fb45680f..00dbde8f 100644
--- a/modules/core/ingest-core/pom.xml
+++ b/modules/core/ingest-core/pom.xml
@@ -20,6 +20,10 @@
com.lingniu.ingest
ingest-codec-common
+
+ com.lingniu.ingest
+ ingest-facts
+
org.springframework.boot
spring-boot-starter
diff --git a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java
index bbf5316c..8ec4412b 100644
--- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java
+++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java
@@ -10,12 +10,17 @@ import com.lingniu.ingest.api.pipeline.RawFrame;
import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor;
import com.lingniu.ingest.core.concurrency.DisruptorEventBus;
import com.lingniu.ingest.core.pipeline.InterceptorChain;
+import com.lingniu.ingest.facts.FactIds;
+import com.lingniu.ingest.facts.VehicleKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
+import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -178,10 +183,15 @@ public final class Dispatcher {
: new LinkedHashMap<>(frame.sourceMeta());
String parsedJson = parsedJson(frame.payload(), meta.remove(RawArchiveKeys.META_PARSED_JSON));
String vin = meta.getOrDefault("vin", "");
+ String phone = meta.getOrDefault("phone", "");
Instant ingestTime = frame.receivedAt() != null ? frame.receivedAt() : Instant.now();
String eventId = nextRawArchiveEventId(ingestTime);
String key = RawArchiveKeys.key(ingestTime, frame.protocolId(), vin, eventId);
- Map archiveMeta = addRawArchiveMetadata(meta, eventId, key);
+ String checksum = "sha256:" + sha256(bytes);
+ String vehicleKey = VehicleKey.derive(frame.protocolId(), vin, phone, eventId);
+ String frameId = FactIds.rawFrameId(
+ frame.protocolId(), vehicleKey, frame.command(), frame.infoType(), ingestTime, checksum);
+ Map archiveMeta = addRawArchiveMetadata(meta, eventId, key, frameId);
VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
eventId,
@@ -196,7 +206,7 @@ public final class Dispatcher {
bytes,
parsedJson);
out.add(raw);
- return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key));
+ return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key), frameId);
}
private static String parsedJson(Object payload, String explicitParsedJson) {
@@ -221,6 +231,15 @@ public final class Dispatcher {
return value.replace("\\", "\\\\").replace("\"", "\\\"");
}
+ private static String sha256(byte[] value) {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ return HexFormat.of().formatHex(digest.digest(value == null ? new byte[0] : value));
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 is unavailable", e);
+ }
+ }
+
private static String nextRawArchiveEventId(Instant ingestTime) {
// 文件名需要可排序且单进程内唯一：毫秒时间放大到微秒量级，再用 AtomicLong 递增兜底。
long base = (ingestTime == null ? Instant.now() : ingestTime).toEpochMilli() * 1000L;
@@ -230,11 +249,16 @@ public final class Dispatcher {
private static Map addRawArchiveMetadata(Map metadata,
String eventId,
- String key) {
+ String key,
+ String frameId) {
Map out = new LinkedHashMap<>(metadata == null ? Map.of() : metadata);
out.putIfAbsent(RawArchiveKeys.META_EVENT_ID, eventId);
out.putIfAbsent(RawArchiveKeys.META_KEY, key);
out.putIfAbsent(RawArchiveKeys.META_URI, RawArchiveKeys.logicalUri(key));
+ if (frameId != null && !frameId.isBlank()) {
+ out.putIfAbsent("rawFrameId", frameId);
+ out.putIfAbsent("frame_id", frameId);
+ }
return Map.copyOf(out);
}
@@ -243,7 +267,8 @@ public final class Dispatcher {
return event;
}
// 派生事件携带 rawArchiveUri，便于后续服务从 Kafka 事件回查原始报文。
- Map metadata = addRawArchiveMetadata(event.metadata(), rawArchive.eventId(), rawArchive.key());
+ Map metadata = addRawArchiveMetadata(
+ event.metadata(), rawArchive.eventId(), rawArchive.key(), rawArchive.frameId());
return switch (event) {
case VehicleEvent.Realtime e -> new VehicleEvent.Realtime(
e.eventId(), e.vin(), e.source(), e.eventTime(), e.ingestTime(),
@@ -273,9 +298,9 @@ public final class Dispatcher {
};
}
- private record RawArchiveLookup(String eventId, String key, String uri) {
+ private record RawArchiveLookup(String eventId, String key, String uri, String frameId) {
private static RawArchiveLookup empty() {
- return new RawArchiveLookup("", "", "");
+ return new RawArchiveLookup("", "", "", "");
}
private boolean isEmpty() {
diff --git a/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherRawArchiveMetadataTest.java b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherRawArchiveMetadataTest.java
index 34529313..1b446ccd 100644
--- a/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherRawArchiveMetadataTest.java
+++ b/modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherRawArchiveMetadataTest.java
@@ -51,15 +51,20 @@ class DispatcherRawArchiveMetadataTest {
VehicleEvent.Location location = awaitEvent(sink, VehicleEvent.Location.class);
String rawArchiveKey = raw.metadata().get("rawArchiveKey");
+ String rawFrameId = raw.metadata().get("rawFrameId");
assertThat(rawArchiveKey).isNotBlank();
+ assertThat(rawFrameId).startsWith("rf_");
assertThat(raw.parsedJson()).contains("\"name\":\"payload-0200\"");
assertThat(raw.metadata()).doesNotContainKey("rawArchiveParsedJson");
assertThat(raw.eventId()).containsOnlyDigits();
assertThat(rawArchiveKey).endsWith("/" + raw.eventId() + ".bin");
+ assertThat(raw.metadata()).containsEntry("frame_id", rawFrameId);
assertThat(location.metadata())
.containsEntry("rawArchiveEventId", raw.eventId())
.containsEntry("rawArchiveKey", rawArchiveKey)
- .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey);
+ .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey)
+ .containsEntry("rawFrameId", rawFrameId)
+ .containsEntry("frame_id", rawFrameId);
} finally {
batchExecutor.close();
eventBus.close();
@@ -92,10 +97,14 @@ class DispatcherRawArchiveMetadataTest {
VehicleEvent.Location location = awaitLocationEvent(sink, "async-location-1");
String rawArchiveKey = raw.metadata().get("rawArchiveKey");
+ String rawFrameId = raw.metadata().get("rawFrameId");
+ assertThat(rawFrameId).startsWith("rf_");
assertThat(location.metadata())
.containsEntry("rawArchiveEventId", raw.eventId())
.containsEntry("rawArchiveKey", rawArchiveKey)
- .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey);
+ .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey)
+ .containsEntry("rawFrameId", rawFrameId)
+ .containsEntry("frame_id", rawFrameId);
} finally {
batchExecutor.close();
eventBus.close();