diff --git a/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveEventSink.java b/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveEventSink.java new file mode 100644 index 00000000..adf5b5a3 --- /dev/null +++ b/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveEventSink.java @@ -0,0 +1,82 @@ +package com.lingniu.ingest.sink.archive; + +import com.lingniu.ingest.api.event.VehicleEvent; +import com.lingniu.ingest.api.sink.EventSink; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayInputStream; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.concurrent.CompletableFuture; + +/** + * 原始报文冷存 Sink:消费 {@link VehicleEvent.RawArchive},把原始入站字节写入 + * {@link ArchiveStore},实现"一帧一落盘"的可回放目标。 + * + *
key 组装:{@code yyyy/MM/dd/ 写操作本身是同步阻塞的本地/对象存储 IO,外层由 Disruptor 的虚拟线程消费者驱动,
+ * 不阻塞平台线程。失败会标在返回的 {@link CompletableFuture} 上,交由 EventBus 打 WARN。
+ */
+public final class ArchiveEventSink implements EventSink {
+
+ private static final Logger log = LoggerFactory.getLogger(ArchiveEventSink.class);
+
+ private static final DateTimeFormatter DATE_KEY =
+ DateTimeFormatter.ofPattern("yyyy/MM/dd").withZone(ZoneOffset.UTC);
+
+ private final ArchiveStore store;
+
+ public ArchiveEventSink(ArchiveStore store) {
+ this.store = store;
+ }
+
+ @Override
+ public String name() {
+ return "archive";
+ }
+
+ @Override
+ public boolean accepts(VehicleEvent event) {
+ return event instanceof VehicleEvent.RawArchive;
+ }
+
+ @Override
+ public CompletableFuture
+ *
+ */
+class ArchiveEventSinkTest {
+
+ private static final DateTimeFormatter DATE_KEY =
+ DateTimeFormatter.ofPattern("yyyy/MM/dd").withZone(ZoneOffset.UTC);
+
+ @Test
+ void acceptsOnlyRawArchiveEvent(@TempDir Path root) {
+ LocalArchiveStore store = new LocalArchiveStore(root.toString());
+ ArchiveEventSink sink = new ArchiveEventSink(store);
+
+ VehicleEvent.RawArchive raw = newRawArchive("LTESTVIN000000001", new byte[]{0x23, 0x23, 0x02});
+ VehicleEvent.Heartbeat hb = new VehicleEvent.Heartbeat(
+ UUID.randomUUID().toString(),
+ "LTESTVIN000000001",
+ ProtocolId.GB32960,
+ Instant.now(),
+ Instant.now(),
+ null,
+ Map.of());
+
+ assertThat(sink.accepts(raw)).isTrue();
+ assertThat(sink.accepts(hb)).isFalse();
+ }
+
+ @Test
+ void publishWritesRawBytesToStableKey(@TempDir Path root) throws Exception {
+ LocalArchiveStore store = new LocalArchiveStore(root.toString());
+ ArchiveEventSink sink = new ArchiveEventSink(store);
+
+ Instant t = Instant.parse("2026-04-20T10:11:12Z");
+ byte[] bytes = new byte[]{0x23, 0x23, 0x02, 0x01, (byte) 0xFE};
+ VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
+ "evt-abc",
+ "LTESTVIN000000001",
+ ProtocolId.GB32960,
+ t,
+ t,
+ null,
+ Map.of(),
+ 0x02,
+ 0,
+ bytes);
+
+ sink.publish(raw).get();
+
+ String expectedKey = DATE_KEY.format(t) + "/GB32960/LTESTVIN000000001/evt-abc.bin";
+ Path expectedFile = root.resolve(expectedKey);
+ assertThat(expectedFile).exists();
+ assertThat(Files.readAllBytes(expectedFile)).containsExactly(bytes);
+ }
+
+ @Test
+ void publishIsIdempotentOnSameKey(@TempDir Path root) throws Exception {
+ LocalArchiveStore store = new LocalArchiveStore(root.toString());
+ ArchiveEventSink sink = new ArchiveEventSink(store);
+
+ byte[] first = new byte[]{0x01, 0x02};
+ byte[] second = new byte[]{0x03, 0x04};
+ Instant t = Instant.parse("2026-04-20T10:11:12Z");
+ VehicleEvent.RawArchive e1 = new VehicleEvent.RawArchive(
+ "evt-same", "VIN", ProtocolId.GB32960, t, t, null, Map.of(), 0, 0, first);
+ VehicleEvent.RawArchive e2 = new VehicleEvent.RawArchive(
+ "evt-same", "VIN", ProtocolId.GB32960, t, t, null, Map.of(), 0, 0, second);
+
+ sink.publish(e1).get();
+ sink.publish(e2).get();
+
+ String key = DATE_KEY.format(t) + "/GB32960/VIN/evt-same.bin";
+ Path file = root.resolve(key);
+ // 约定:同一 eventId 路径重复写时直接覆盖(put 语义),保留最新一份。
+ assertThat(Files.readAllBytes(file)).containsExactly(second);
+ }
+
+ private static VehicleEvent.RawArchive newRawArchive(String vin, byte[] bytes) {
+ return new VehicleEvent.RawArchive(
+ UUID.randomUUID().toString(),
+ vin,
+ ProtocolId.GB32960,
+ Instant.now(),
+ Instant.now(),
+ null,
+ Map.of(),
+ 0,
+ 0,
+ bytes);
+ }
+}