From 019512bac24c133a86955975a77a8676ce97944f Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 23 Jun 2026 15:42:35 +0800 Subject: [PATCH] fix: restore sink archive implementation --- .../ingest/sink/archive/ArchiveStore.java | 17 ++++ .../sink/archive/LocalArchiveStore.java | 88 +++++++++++++++++++ .../sink/archive/RawArchiveEventSink.java | 41 +++++++++ .../config/SinkArchiveAutoConfiguration.java | 33 +++++++ .../archive/config/SinkArchiveProperties.java | 35 ++++++++ 5 files changed, 214 insertions(+) create mode 100644 modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveStore.java create mode 100644 modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/LocalArchiveStore.java create mode 100644 modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java create mode 100644 modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java create mode 100644 modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveStore.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveStore.java new file mode 100644 index 00000000..211d633c --- /dev/null +++ b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/ArchiveStore.java @@ -0,0 +1,17 @@ +package com.lingniu.ingest.sink.archive; + +import java.io.IOException; +import java.io.InputStream; + +public interface ArchiveStore { + + String put(String key, InputStream data, long length) throws IOException; + + String append(String key, byte[] chunk) throws IOException; + + InputStream get(String key) throws IOException; + + boolean exists(String key); + + long size(String key) throws IOException; +} diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/LocalArchiveStore.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/LocalArchiveStore.java new file mode 100644 index 00000000..e220920b --- /dev/null +++ b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/LocalArchiveStore.java @@ -0,0 +1,88 @@ +package com.lingniu.ingest.sink.archive; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.URI; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; + +public final class LocalArchiveStore implements ArchiveStore { + + private final Path root; + + public LocalArchiveStore(String root) { + this(rootPath(root)); + } + + public LocalArchiveStore(Path root) { + this.root = root.toAbsolutePath().normalize(); + } + + @Override + public String put(String key, InputStream data, long length) throws IOException { + if (data == null) { + throw new IllegalArgumentException("data must not be null"); + } + Path target = resolve(key); + Files.createDirectories(target.getParent()); + try (OutputStream out = Files.newOutputStream( + target, + StandardOpenOption.CREATE, + StandardOpenOption.TRUNCATE_EXISTING, + StandardOpenOption.WRITE)) { + data.transferTo(out); + } + return target.toUri().toString(); + } + + @Override + public String append(String key, byte[] chunk) throws IOException { + Path target = resolve(key); + Files.createDirectories(target.getParent()); + Files.write( + target, + chunk == null ? new byte[0] : chunk, + StandardOpenOption.CREATE, + StandardOpenOption.APPEND, + StandardOpenOption.WRITE); + return target.toUri().toString(); + } + + @Override + public InputStream get(String key) throws IOException { + return Files.newInputStream(resolve(key), StandardOpenOption.READ); + } + + @Override + public boolean exists(String key) { + return Files.exists(resolve(key)); + } + + @Override + public long size(String key) throws IOException { + return Files.size(resolve(key)); + } + + private Path resolve(String key) { + if (key == null || key.isBlank()) { + throw new IllegalArgumentException("archive key must not be blank"); + } + Path target = root.resolve(key).normalize(); + if (!target.startsWith(root)) { + throw new IllegalArgumentException("archive key escapes root: " + key); + } + return target; + } + + private static Path rootPath(String value) { + if (value == null || value.isBlank()) { + return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive"); + } + if (value.startsWith("file://")) { + return Path.of(URI.create(value)); + } + return Path.of(value); + } +} diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java new file mode 100644 index 00000000..06af1305 --- /dev/null +++ b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java @@ -0,0 +1,41 @@ +package com.lingniu.ingest.sink.archive; + +import com.lingniu.ingest.api.event.RawArchiveKeys; +import com.lingniu.ingest.api.event.VehicleEvent; +import com.lingniu.ingest.api.sink.EventSink; + +import java.io.ByteArrayInputStream; +import java.util.concurrent.CompletableFuture; + +public final class RawArchiveEventSink implements EventSink { + + private final ArchiveStore store; + + public RawArchiveEventSink(ArchiveStore store) { + this.store = store; + } + + @Override + public String name() { + return "raw-archive"; + } + + @Override + public CompletableFuture publish(VehicleEvent event) { + if (!(event instanceof VehicleEvent.RawArchive raw)) { + return CompletableFuture.completedFuture(null); + } + byte[] bytes = raw.rawBytes() == null ? new byte[0] : raw.rawBytes(); + try { + store.put(RawArchiveKeys.key(raw), new ByteArrayInputStream(bytes), bytes.length); + return CompletableFuture.completedFuture(null); + } catch (Exception e) { + return CompletableFuture.failedFuture(e); + } + } + + @Override + public boolean accepts(VehicleEvent event) { + return event instanceof VehicleEvent.RawArchive; + } +} diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java new file mode 100644 index 00000000..73efc1d9 --- /dev/null +++ b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java @@ -0,0 +1,33 @@ +package com.lingniu.ingest.sink.archive.config; + +import com.lingniu.ingest.sink.archive.ArchiveStore; +import com.lingniu.ingest.sink.archive.LocalArchiveStore; +import com.lingniu.ingest.sink.archive.RawArchiveEventSink; +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; + +@AutoConfiguration +@EnableConfigurationProperties(SinkArchiveProperties.class) +@ConditionalOnProperty( + prefix = "lingniu.ingest.sink.archive", + name = "enabled", + havingValue = "true", + matchIfMissing = true) +public class SinkArchiveAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty(prefix = "lingniu.ingest.sink.archive", name = "type", havingValue = "local", matchIfMissing = true) + public ArchiveStore archiveStore(SinkArchiveProperties properties) { + return new LocalArchiveStore(properties.getPath()); + } + + @Bean + @ConditionalOnMissingBean + public RawArchiveEventSink rawArchiveEventSink(ArchiveStore store) { + return new RawArchiveEventSink(store); + } +} diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java new file mode 100644 index 00000000..f64e2acb --- /dev/null +++ b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java @@ -0,0 +1,35 @@ +package com.lingniu.ingest.sink.archive.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "lingniu.ingest.sink.archive") +public class SinkArchiveProperties { + + private boolean enabled = true; + private String type = "local"; + private String path = System.getProperty("java.io.tmpdir") + "/lingniu-archive"; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getPath() { + return path; + } + + public void setPath(String path) { + this.path = path; + } +}