From dd838b4f73c56f170cc732ce35226848aad00e5f Mon Sep 17 00:00:00 2001 From: kkfluous Date: Mon, 20 Apr 2026 15:54:54 +0800 Subject: [PATCH] feat(sink-archive): ArchiveEventSink consumes RawArchive events to cold storage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新 EventSink 实现： - accepts 仅 VehicleEvent.RawArchive - publish 写 ArchiveStore，key = yyyy/MM/dd///.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) --- .../ingest/sink/archive/ArchiveEventSink.java | 82 +++++++++++++ .../config/SinkArchiveAutoConfiguration.java | 15 +++ .../sink/archive/ArchiveEventSinkTest.java | 114 ++++++++++++++++++ 3 files changed, 211 insertions(+) create mode 100644 sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveEventSink.java create mode 100644 sink-archive/src/test/java/com/lingniu/ingest/sink/archive/ArchiveEventSinkTest.java 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///.bin} + * （基于 {@link VehicleEvent#ingestTime()} 的 UTC 日期）。相同 eventId 调两次会覆盖前次写入。 + * + *

写操作本身是同步阻塞的本地/对象存储 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 publish(VehicleEvent event) { + CompletableFuture 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"; + } +} diff --git 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 index f455b55e..a5346255 100644 --- 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); + } } diff --git 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 new file mode 100644 index 00000000..33c94d78 --- /dev/null +++ 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} 单元测试： + * + *

    + *
  • 只接受 {@link VehicleEvent.RawArchive}，其它事件通过 {@link + * ArchiveEventSink#accepts(VehicleEvent)} 拒绝 + *
  • publish 成功将原始字节写到 ArchiveStore 的稳定 key 路径下 + *
  • key 包含日期分片 + 协议名 + VIN + eventId，便于按车按天回放 + *
+ */ +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); + } +}