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 8ec4412b..e066ee67 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,17 +10,13 @@ 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 com.lingniu.ingest.facts.RawFrameIdentity; 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; @@ -187,11 +183,9 @@ public final class Dispatcher { Instant ingestTime = frame.receivedAt() != null ? frame.receivedAt() : Instant.now(); String eventId = nextRawArchiveEventId(ingestTime); String key = RawArchiveKeys.key(ingestTime, frame.protocolId(), vin, eventId); - 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); + RawFrameIdentity identity = RawFrameIdentity.derive( + frame.protocolId(), vin, phone, eventId, frame.command(), frame.infoType(), ingestTime, bytes); + Map archiveMeta = addRawArchiveMetadata(meta, eventId, key, identity.frameId()); VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive( eventId, @@ -206,7 +200,7 @@ public final class Dispatcher { bytes, parsedJson); out.add(raw); - return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key), frameId); + return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key), identity.frameId()); } private static String parsedJson(Object payload, String explicitParsedJson) { @@ -231,15 +225,6 @@ 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; diff --git a/modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/RawFrameIdentity.java b/modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/RawFrameIdentity.java new file mode 100644 index 00000000..58b9bfc5 --- /dev/null +++ b/modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/RawFrameIdentity.java @@ -0,0 +1,43 @@ +package com.lingniu.ingest.facts; + +import com.lingniu.ingest.api.ProtocolId; + +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.Instant; +import java.util.HexFormat; + +public record RawFrameIdentity( + String checksum, + String vehicleKey, + String frameId +) { + public RawFrameIdentity { + checksum = checksum == null ? "" : checksum; + vehicleKey = vehicleKey == null ? "" : vehicleKey; + frameId = frameId == null ? "" : frameId; + } + + public static RawFrameIdentity derive(ProtocolId protocol, + String vin, + String phone, + String rawIdentitySeed, + int messageId, + int subType, + Instant receivedAt, + byte[] rawBytes) { + String checksum = "sha256:" + sha256(rawBytes); + String vehicleKey = VehicleKey.derive(protocol, vin, phone, rawIdentitySeed); + String frameId = FactIds.rawFrameId(protocol, vehicleKey, messageId, subType, receivedAt, checksum); + return new RawFrameIdentity(checksum, vehicleKey, frameId); + } + + 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); + } + } +} diff --git a/modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/RawFrameIdentityTest.java b/modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/RawFrameIdentityTest.java new file mode 100644 index 00000000..c0a59f1d --- /dev/null +++ b/modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/RawFrameIdentityTest.java @@ -0,0 +1,36 @@ +package com.lingniu.ingest.facts; + +import com.lingniu.ingest.api.ProtocolId; +import org.junit.jupiter.api.Test; + +import java.time.Instant; + +import static org.assertj.core.api.Assertions.assertThat; + +class RawFrameIdentityTest { + + @Test + void derivesChecksumVehicleKeyAndFrameIdTogether() { + RawFrameIdentity identity = RawFrameIdentity.derive( + ProtocolId.JT808, + "", + "13307795425", + "1782900000000000", + 0x0200, + 0, + Instant.parse("2026-07-01T10:00:00Z"), + new byte[]{0x7e, 0x02, 0x00, 0x7e}); + + assertThat(identity.checksum()).startsWith("sha256:"); + assertThat(identity.vehicleKey()).isEqualTo("jt808:13307795425"); + assertThat(identity.frameId()) + .isEqualTo(FactIds.rawFrameId( + ProtocolId.JT808, + "jt808:13307795425", + 0x0200, + 0, + Instant.parse("2026-07-01T10:00:00Z"), + identity.checksum())) + .startsWith("rf_"); + } +} diff --git a/modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/EnvelopeMapper.java b/modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/EnvelopeMapper.java index 405da958..04cf415d 100644 --- a/modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/EnvelopeMapper.java +++ b/modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/EnvelopeMapper.java @@ -8,17 +8,12 @@ import com.lingniu.ingest.api.event.TelemetryFieldValue; import com.lingniu.ingest.api.event.TelemetrySnapshot; import com.lingniu.ingest.api.event.VehicleEvent; import com.lingniu.ingest.api.event.VehicleEventTelemetrySnapshotMapper; -import com.lingniu.ingest.facts.FactIds; -import com.lingniu.ingest.facts.VehicleKey; +import com.lingniu.ingest.facts.RawFrameIdentity; import com.lingniu.ingest.sink.kafka.proto.ParseStatusProto; import com.lingniu.ingest.sink.kafka.proto.TelemetryField; import com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot.Builder; import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.util.HexFormat; - /** * 领域事件 → Protobuf Envelope 的纯函数映射。通过 switch 模式匹配穷尽处理， * 新增 {@link VehicleEvent} 子类型时编译期就会提示补齐分支。 @@ -80,29 +75,28 @@ public final class EnvelopeMapper { String key = rawArchiveKey(ra); String uri = rawArchiveUri(ra, key); String phone = metadataValue(ra, "phone"); - String vehicleKey = VehicleKey.derive(ra.source(), ra.vin(), phone, ra.eventId()); - String checksum = sha256(rawBytes); - String frameId = FactIds.rawFrameId( - ra.source(), vehicleKey, ra.command(), ra.infoType(), ra.ingestTime(), checksum); + RawFrameIdentity identity = RawFrameIdentity.derive( + ra.source(), ra.vin(), phone, ra.eventId(), ra.command(), ra.infoType(), + ra.ingestTime(), rawBytes); // RAW envelope 只携带 URI/size 引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。 b.putMetadata(RawArchiveKeys.META_KEY, key); b.putMetadata(RawArchiveKeys.META_URI, uri); b.putMetadata(RawArchiveKeys.META_EVENT_ID, ra.eventId()); b.setRawArchive(com.lingniu.ingest.sink.kafka.proto.RawArchiveRef.newBuilder() .setUri(uri) - .setChecksum(checksum) + .setChecksum(identity.checksum()) .setSizeBytes(size) .setParsedJson(nullToEmpty(ra.parsedJson())) .build()); b.setRawFrameFact(com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload.newBuilder() - .setFrameId(frameId) - .setVehicleKey(vehicleKey) + .setFrameId(identity.frameId()) + .setVehicleKey(identity.vehicleKey()) .setVin(nullToEmpty(ra.vin())) .setPhone(phone) .setMessageId(ra.command()) .setSubType(ra.infoType()) .setRawUri(uri) - .setChecksum(checksum) + .setChecksum(identity.checksum()) .setRawSizeBytes(size) .setParseStatus(parseStatus(ra)) .setParseError(parseError(ra)) @@ -229,12 +223,4 @@ public final class EnvelopeMapper { return Boolean.parseBoolean(metadataValue(event, key)); } - private static String sha256(byte[] bytes) { - try { - MessageDigest digest = MessageDigest.getInstance("SHA-256"); - return "sha256:" + HexFormat.of().formatHex(digest.digest(bytes == null ? new byte[0] : bytes)); - } catch (NoSuchAlgorithmException e) { - throw new IllegalStateException("SHA-256 is unavailable", e); - } - } }