feat(sink-archive): ArchiveEventSink consumes RawArchive events to cold storage

新 EventSink 实现： - accepts 仅 VehicleEvent.RawArchive - publish 写 ArchiveStore，key = yyyy/MM/dd/<source>/<vin>/<eventId>.bin （基于 ingestTime 的 UTC 日期分片，便于按车按天回放） - 相同 eventId 重复写按 put 语义覆盖 - 写失败在返回的 CompletableFuture 上 completeExceptionally，EventBus 打 WARN AutoConfig 在 ArchiveStore bean 存在时装配 ArchiveEventSink，Disruptor EventBus 启动时自动把它纳入扇出 handler 列表。Disruptor 使用虚拟线程消费，本地/S3/OSS 的 blocking IO 不会 pin 平台线程。单测覆盖 accepts 过滤、稳定 key 写入、幂等覆盖三种情形。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-20 15:54:54 +08:00
parent bfc3c29c40
commit dd838b4f73
3 changed files with 211 additions and 0 deletions
--- 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
@@ -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}，实现"一帧一落盘"的可回放目标。
+ *
+ * <p>key 组装：{@code yyyy/MM/dd/<source>/<vin>/<eventId>.bin}
+ * （基于 {@link VehicleEvent#ingestTime()} 的 UTC 日期）。相同 eventId 调两次会覆盖前次写入。
+ *
+ * <p>写操作本身是同步阻塞的本地/对象存储 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<Void> publish(VehicleEvent event) {
+        CompletableFuture<Void> cf = new CompletableFuture<>();
+        if (!(event instanceof VehicleEvent.RawArchive raw)) {
+            // 理论上 accepts 过滤后不会进入；保险起见快速返回避免误落盘。
+            cf.complete(null);
+            return cf;
+        }
+        byte[] bytes = raw.rawBytes();
+        if (bytes == null || bytes.length == 0) {
+            cf.complete(null);
+            return cf;
+        }
+        String key = buildKey(raw);
+        try {
+            String uri = store.put(key, new ByteArrayInputStream(bytes), bytes.length);
+            if (log.isDebugEnabled()) {
+                log.debug("archive put vin={} eventId={} size={} uri={}",
+                        raw.vin(), raw.eventId(), bytes.length, uri);
+            }
+            cf.complete(null);
+        } catch (Exception e) {
+            log.warn("archive put failed vin={} eventId={} key={}",
+                    raw.vin(), raw.eventId(), key, e);
+            cf.completeExceptionally(e);
+        }
+        return cf;
+    }
+
+    private static String buildKey(VehicleEvent.RawArchive raw) {
+        String dateKey = DATE_KEY.format(raw.ingestTime());
+        String vin = raw.vin() == null || raw.vin().isBlank() ? "unknown-vin" : raw.vin();
+        String eventId = raw.eventId() == null || raw.eventId().isBlank()
+                ? Long.toHexString(System.nanoTime()) : raw.eventId();
+        return dateKey + "/" + raw.source().name() + "/" + vin + "/" + eventId + ".bin";
+    }
+}
--- a/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java
+++ b/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java
@@ -1,8 +1,10 @@
 package com.lingniu.ingest.sink.archive.config;

+import com.lingniu.ingest.sink.archive.ArchiveEventSink;
 import com.lingniu.ingest.sink.archive.ArchiveStore;
 import com.lingniu.ingest.sink.archive.LocalArchiveStore;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -23,4 +25,17 @@ public class SinkArchiveAutoConfiguration {
                            + " (only 'local' is implemented; s3/oss TBD)");
        };
    }
+
+    /**
+     * 注册 {@link ArchiveEventSink} 为 EventBus 的一个出口，消费 {@link
+     * com.lingniu.ingest.api.event.VehicleEvent.RawArchive} 事件。仅在存在
+     * {@link ArchiveStore} bean 时装配；若未启用 sink-archive 或 archive 配置缺失，
+     * 则不落地原始字节，但 RawArchive 事件本身仍会被 Disruptor 丢弃（无消费者）。
+     */
+    @Bean
+    @ConditionalOnBean(ArchiveStore.class)
+    @ConditionalOnMissingBean
+    public ArchiveEventSink archiveEventSink(ArchiveStore store) {
+        return new ArchiveEventSink(store);
+    }
 }
--- a/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/ArchiveEventSinkTest.java
+++ b/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/ArchiveEventSinkTest.java
@@ -0,0 +1,114 @@
+package com.lingniu.ingest.sink.archive;
+
+import com.lingniu.ingest.api.ProtocolId;
+import com.lingniu.ingest.api.event.VehicleEvent;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * {@link ArchiveEventSink} 单元测试：
+ *
+ * <ul>
+ *   <li>只接受 {@link VehicleEvent.RawArchive}，其它事件通过 {@link
+ *       ArchiveEventSink#accepts(VehicleEvent)} 拒绝
+ *   <li>publish 成功将原始字节写到 ArchiveStore 的稳定 key 路径下
+ *   <li>key 包含日期分片 + 协议名 + VIN + eventId，便于按车按天回放
+ * </ul>
+ */
+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);
+    }
+}