diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java new file mode 100644 index 00000000..91fcd539 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueries.java @@ -0,0 +1,65 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.util.ArrayList; +import java.util.List; + +public final class TdengineHistoryQueries { + + private static final String RAW_COLUMNS = "ts, frame_id, received_at, message_id, sub_type, event_time, " + + "raw_uri, checksum, raw_size_bytes, parse_status, parse_error, peer, metadata_json, " + + "protocol, vehicle_key, vin, phone"; + private static final String LOCATION_COLUMNS = "ts, fact_id, frame_id, received_at, longitude, latitude, " + + "altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, raw_uri, " + + "metadata_json, protocol, vehicle_key, vin, phone"; + + private final TdengineHistorySchema schema; + + public TdengineHistoryQueries(TdengineHistorySchema schema) { + if (schema == null) { + throw new IllegalArgumentException("schema must not be null"); + } + this.schema = schema; + } + + public TdengineQueryStatement rawFrames(TdengineRawFrameQuery query) { + String table = schema.rawFrameTable(query.protocol(), query.vehicleKey()); + return query(table, RAW_COLUMNS, "frame_id", query.from(), query.to(), + query.order(), query.limit(), query.cursor()); + } + + public TdengineQueryStatement locations(TdengineLocationQuery query) { + String table = schema.locationTable(query.protocol(), query.vehicleKey()); + return query(table, LOCATION_COLUMNS, "fact_id", query.from(), query.to(), + query.order(), query.limit(), query.cursor()); + } + + private static TdengineQueryStatement query(String table, + String columns, + String tieBreakerColumn, + Object from, + Object to, + TdengineQueryOrder order, + int limit, + TdenginePageCursor cursor) { + List values = new ArrayList<>(); + StringBuilder sql = new StringBuilder(256) + .append("SELECT ").append(columns) + .append(" FROM ").append(table) + .append(" WHERE ts >= ? AND ts < ?"); + values.add(from); + values.add(to); + if (cursor != null) { + String op = order == TdengineQueryOrder.ASC ? ">" : "<"; + sql.append(" AND (ts ").append(op).append(" ? OR (ts = ? AND ") + .append(tieBreakerColumn).append(' ').append(op).append(" ?))"); + values.add(cursor.ts()); + values.add(cursor.ts()); + values.add(cursor.tieBreaker()); + } + sql.append(" ORDER BY ts ").append(order.name()) + .append(", ").append(tieBreakerColumn).append(' ').append(order.name()) + .append(" LIMIT ?"); + values.add(limit); + return new TdengineQueryStatement(sql.toString(), values); + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java new file mode 100644 index 00000000..bd99fe11 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryReader.java @@ -0,0 +1,10 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.io.IOException; + +public interface TdengineHistoryReader { + + TdenginePage queryRawFrames(TdengineRawFrameQuery query) throws IOException; + + TdenginePage queryLocations(TdengineLocationQuery query) throws IOException; +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java new file mode 100644 index 00000000..974be5ca --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReader.java @@ -0,0 +1,138 @@ +package com.lingniu.ingest.tdenginehistory; + +import javax.sql.DataSource; +import java.io.IOException; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.function.Function; + +public final class TdengineJdbcHistoryReader implements TdengineHistoryReader { + + private final DataSource dataSource; + private final TdengineHistoryQueries queries; + + public TdengineJdbcHistoryReader(DataSource dataSource, TdengineHistorySchema schema) { + if (dataSource == null) { + throw new IllegalArgumentException("dataSource must not be null"); + } + this.dataSource = dataSource; + this.queries = new TdengineHistoryQueries(schema); + } + + @Override + public TdenginePage queryRawFrames(TdengineRawFrameQuery query) throws IOException { + int pageSize = Math.min(query.limit(), 1000); + TdengineQueryStatement statement = queries.rawFrames(query.withLimit(pageSize + 1)); + return readPage(statement, pageSize, this::rawFrame, row -> new TdenginePageCursor(row.ts(), row.frameId())); + } + + @Override + public TdenginePage queryLocations(TdengineLocationQuery query) throws IOException { + int pageSize = Math.min(query.limit(), 1000); + TdengineQueryStatement statement = queries.locations(query.withLimit(pageSize + 1)); + return readPage(statement, pageSize, this::location, row -> new TdenginePageCursor(row.ts(), row.factId())); + } + + private TdenginePage readPage(TdengineQueryStatement statement, + int pageSize, + SqlRowMapper mapper, + Function cursor) throws IOException { + List rows = new ArrayList<>(); + try (Connection connection = dataSource.getConnection(); + PreparedStatement prepared = connection.prepareStatement(statement.sql())) { + bind(prepared, statement.values()); + try (ResultSet resultSet = prepared.executeQuery()) { + while (resultSet.next()) { + rows.add(mapper.map(resultSet)); + } + } + } catch (SQLException e) { + throw new IOException("tdengine history query failed", e); + } + boolean hasMore = rows.size() > pageSize; + List items = hasMore ? List.copyOf(rows.subList(0, pageSize)) : List.copyOf(rows); + Optional next = hasMore && !items.isEmpty() + ? Optional.of(cursor.apply(items.getLast())) + : Optional.empty(); + return new TdenginePage<>(items, next); + } + + private static void bind(PreparedStatement prepared, List values) throws SQLException { + for (int i = 0; i < values.size(); i++) { + Object value = values.get(i); + int parameterIndex = i + 1; + if (value instanceof Instant instant) { + prepared.setTimestamp(parameterIndex, Timestamp.from(instant)); + } else { + prepared.setObject(parameterIndex, value); + } + } + } + + private TdengineRawFrameRow rawFrame(ResultSet rs) throws SQLException { + return new TdengineRawFrameRow( + instant(rs, "ts"), + rs.getString("frame_id"), + instant(rs, "received_at"), + rs.getInt("message_id"), + rs.getInt("sub_type"), + instant(rs, "event_time"), + rs.getString("raw_uri"), + rs.getString("checksum"), + rs.getLong("raw_size_bytes"), + rs.getString("parse_status"), + rs.getString("parse_error"), + rs.getString("peer"), + rs.getString("metadata_json"), + rs.getString("protocol"), + rs.getString("vehicle_key"), + rs.getString("vin"), + rs.getString("phone") + ); + } + + private TdengineLocationRow location(ResultSet rs) throws SQLException { + return new TdengineLocationRow( + instant(rs, "ts"), + rs.getString("fact_id"), + rs.getString("frame_id"), + instant(rs, "received_at"), + rs.getDouble("longitude"), + rs.getDouble("latitude"), + rs.getDouble("altitude_m"), + rs.getDouble("speed_kmh"), + rs.getDouble("direction_deg"), + rs.getLong("alarm_flag"), + rs.getLong("status_flag"), + nullableDouble(rs, "total_mileage_km"), + rs.getString("raw_uri"), + rs.getString("metadata_json"), + rs.getString("protocol"), + rs.getString("vehicle_key"), + rs.getString("vin"), + rs.getString("phone") + ); + } + + private static Instant instant(ResultSet rs, String column) throws SQLException { + Timestamp timestamp = rs.getTimestamp(column); + return timestamp == null ? null : timestamp.toInstant(); + } + + private static Double nullableDouble(ResultSet rs, String column) throws SQLException { + Object value = rs.getObject(column); + return value instanceof Number number ? number.doubleValue() : null; + } + + @FunctionalInterface + private interface SqlRowMapper { + T map(ResultSet rs) throws SQLException; + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineLocationQuery.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineLocationQuery.java new file mode 100644 index 00000000..700484f8 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineLocationQuery.java @@ -0,0 +1,31 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.time.Instant; + +public record TdengineLocationQuery( + String protocol, + String vehicleKey, + Instant from, + Instant to, + TdengineQueryOrder order, + int limit, + TdenginePageCursor cursor +) { + public TdengineLocationQuery { + if (protocol == null || protocol.isBlank()) { + throw new IllegalArgumentException("protocol must not be blank"); + } + if (vehicleKey == null || vehicleKey.isBlank()) { + throw new IllegalArgumentException("vehicleKey must not be blank"); + } + if (from == null || to == null || !from.isBefore(to)) { + throw new IllegalArgumentException("query time range must be valid"); + } + order = order == null ? TdengineQueryOrder.DESC : order; + limit = Math.max(1, Math.min(limit, 1001)); + } + + TdengineLocationQuery withLimit(int newLimit) { + return new TdengineLocationQuery(protocol, vehicleKey, from, to, order, newLimit, cursor); + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdenginePage.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdenginePage.java new file mode 100644 index 00000000..5447cf98 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdenginePage.java @@ -0,0 +1,12 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.util.List; +import java.util.Optional; + +public record TdenginePage(List items, Optional nextCursor) { + + public TdenginePage { + items = items == null ? List.of() : List.copyOf(items); + nextCursor = nextCursor == null ? Optional.empty() : nextCursor; + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdenginePageCursor.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdenginePageCursor.java new file mode 100644 index 00000000..76a2a3e3 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdenginePageCursor.java @@ -0,0 +1,13 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.time.Instant; + +public record TdenginePageCursor(Instant ts, String tieBreaker) { + + public TdenginePageCursor { + if (ts == null) { + throw new IllegalArgumentException("cursor ts must not be null"); + } + tieBreaker = tieBreaker == null ? "" : tieBreaker; + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineQueryOrder.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineQueryOrder.java new file mode 100644 index 00000000..3c1c0e71 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineQueryOrder.java @@ -0,0 +1,6 @@ +package com.lingniu.ingest.tdenginehistory; + +public enum TdengineQueryOrder { + ASC, + DESC +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineQueryStatement.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineQueryStatement.java new file mode 100644 index 00000000..034fffd3 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineQueryStatement.java @@ -0,0 +1,15 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +public record TdengineQueryStatement(String sql, List values) { + + public TdengineQueryStatement { + if (sql == null || sql.isBlank()) { + throw new IllegalArgumentException("sql must not be blank"); + } + values = values == null ? List.of() : Collections.unmodifiableList(new ArrayList<>(values)); + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java new file mode 100644 index 00000000..5bffc337 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameQuery.java @@ -0,0 +1,31 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.time.Instant; + +public record TdengineRawFrameQuery( + String protocol, + String vehicleKey, + Instant from, + Instant to, + TdengineQueryOrder order, + int limit, + TdenginePageCursor cursor +) { + public TdengineRawFrameQuery { + if (protocol == null || protocol.isBlank()) { + throw new IllegalArgumentException("protocol must not be blank"); + } + if (vehicleKey == null || vehicleKey.isBlank()) { + throw new IllegalArgumentException("vehicleKey must not be blank"); + } + if (from == null || to == null || !from.isBefore(to)) { + throw new IllegalArgumentException("query time range must be valid"); + } + order = order == null ? TdengineQueryOrder.DESC : order; + limit = Math.max(1, Math.min(limit, 1001)); + } + + TdengineRawFrameQuery withLimit(int newLimit) { + return new TdengineRawFrameQuery(protocol, vehicleKey, from, to, order, newLimit, cursor); + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java index 519a1156..b808308e 100644 --- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java @@ -2,7 +2,9 @@ package com.lingniu.ingest.tdenginehistory.config; import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema; import com.lingniu.ingest.tdenginehistory.TdengineHistoryStatements; +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter; +import com.lingniu.ingest.tdenginehistory.TdengineJdbcHistoryReader; import com.lingniu.ingest.tdenginehistory.TdengineJdbcHistoryWriter; import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; @@ -47,4 +49,15 @@ public class TdengineHistoryAutoConfiguration { public TdengineHistoryWriter tdengineHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) { return new TdengineJdbcHistoryWriter(dataSource, schema); } + + @Bean + @ConditionalOnMissingBean + @ConditionalOnBean(DataSource.class) + @ConditionalOnProperty( + prefix = "lingniu.ingest.tdengine-history", + name = "enabled", + havingValue = "true") + public TdengineHistoryReader tdengineHistoryReader(DataSource dataSource, TdengineHistorySchema schema) { + return new TdengineJdbcHistoryReader(dataSource, schema); + } } diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java new file mode 100644 index 00000000..918d0364 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryQueriesTest.java @@ -0,0 +1,59 @@ +package com.lingniu.ingest.tdenginehistory; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; + +import static org.assertj.core.api.Assertions.assertThat; + +class TdengineHistoryQueriesTest { + + private final TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history"); + private final TdengineHistoryQueries queries = new TdengineHistoryQueries(schema); + + @Test + void rawFrameQueryUsesVehicleChildTableAndKeysetPagination() { + TdengineRawFrameQuery query = new TdengineRawFrameQuery( + "JT808", + "jt808:g7gps", + Instant.parse("2026-06-29T00:00:00Z"), + Instant.parse("2026-06-30T00:00:00Z"), + TdengineQueryOrder.ASC, + 101, + new TdenginePageCursor(Instant.parse("2026-06-29T01:00:00Z"), "frame-100")); + + TdengineQueryStatement statement = queries.rawFrames(query); + + assertThat(statement.sql()) + .startsWith("SELECT ts, frame_id, received_at, message_id") + .contains(" FROM raw_jt808_") + .contains(" WHERE ts >= ? AND ts < ?") + .contains(" AND (ts > ? OR (ts = ? AND frame_id > ?))") + .contains(" ORDER BY ts ASC, frame_id ASC LIMIT ?"); + assertThat(statement.values()) + .containsExactly( + query.from(), query.to(), + query.cursor().ts(), query.cursor().ts(), query.cursor().tieBreaker(), + 101); + } + + @Test + void locationQueryCapsLimitAndUsesDescendingCursor() { + TdengineLocationQuery query = new TdengineLocationQuery( + "GB32960", + "vin:VIN123", + Instant.parse("2026-06-29T00:00:00Z"), + Instant.parse("2026-06-30T00:00:00Z"), + TdengineQueryOrder.DESC, + 5000, + new TdenginePageCursor(Instant.parse("2026-06-29T20:00:00Z"), "fact-200")); + + TdengineQueryStatement statement = queries.locations(query); + + assertThat(statement.sql()) + .contains(" FROM loc_gb32960_") + .contains(" AND (ts < ? OR (ts = ? AND fact_id < ?))") + .contains(" ORDER BY ts DESC, fact_id DESC LIMIT ?"); + assertThat(statement.values().getLast()).isEqualTo(1001); + } +} diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java new file mode 100644 index 00000000..6aff69cf --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryReaderTest.java @@ -0,0 +1,171 @@ +package com.lingniu.ingest.tdenginehistory; + +import org.junit.jupiter.api.Test; + +import javax.sql.DataSource; +import java.lang.reflect.Proxy; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class TdengineJdbcHistoryReaderTest { + + @Test + void readsRawFramesAndReturnsNextCursorWhenLimitHasMoreRow() throws Exception { + RecordingQueryJdbc jdbc = new RecordingQueryJdbc(); + jdbc.rows.add(rawFrameResult("frame-1", "2026-06-29T05:00:01Z")); + jdbc.rows.add(rawFrameResult("frame-2", "2026-06-29T05:00:02Z")); + TdengineJdbcHistoryReader reader = new TdengineJdbcHistoryReader( + jdbc.dataSource(), new TdengineHistorySchema("vehicle_history")); + + TdenginePage page = reader.queryRawFrames(new TdengineRawFrameQuery( + "JT808", + "jt808:g7gps", + Instant.parse("2026-06-29T00:00:00Z"), + Instant.parse("2026-06-30T00:00:00Z"), + TdengineQueryOrder.ASC, + 1, + null)); + + assertThat(page.items()) + .extracting(TdengineRawFrameRow::frameId) + .containsExactly("frame-1"); + assertThat(page.nextCursor()) + .contains(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "frame-1")); + assertThat(jdbc.preparedSql).contains("LIMIT ?"); + assertThat(jdbc.boundValues.get(3)).isEqualTo(2); + } + + @Test + void readsLocationRowsWithoutNextCursorWhenPageIsComplete() throws Exception { + RecordingQueryJdbc jdbc = new RecordingQueryJdbc(); + jdbc.rows.add(locationResult("fact-1", "2026-06-29T05:00:01Z")); + TdengineJdbcHistoryReader reader = new TdengineJdbcHistoryReader( + jdbc.dataSource(), new TdengineHistorySchema("vehicle_history")); + + TdenginePage page = reader.queryLocations(new TdengineLocationQuery( + "GB32960", + "vin:VIN123", + Instant.parse("2026-06-29T00:00:00Z"), + Instant.parse("2026-06-30T00:00:00Z"), + TdengineQueryOrder.DESC, + 10, + null)); + + assertThat(page.items()) + .extracting(TdengineLocationRow::factId) + .containsExactly("fact-1"); + assertThat(page.nextCursor()).isEmpty(); + assertThat(page.items().getFirst().longitude()).isEqualTo(113.12); + } + + private static Map rawFrameResult(String frameId, String ts) { + Map row = new LinkedHashMap<>(); + row.put("ts", Timestamp.from(Instant.parse(ts))); + row.put("frame_id", frameId); + row.put("received_at", Timestamp.from(Instant.parse(ts).plusMillis(500))); + row.put("message_id", 0x0200); + row.put("sub_type", 0); + row.put("event_time", Timestamp.from(Instant.parse(ts))); + row.put("raw_uri", "archive://jt808/" + frameId + ".bin"); + row.put("checksum", "sha256:" + frameId); + row.put("raw_size_bytes", 67L); + row.put("parse_status", "SUCCEEDED"); + row.put("parse_error", ""); + row.put("peer", "10.0.0.1:808"); + row.put("metadata_json", "{}"); + row.put("protocol", "JT808"); + row.put("vehicle_key", "jt808:g7gps"); + row.put("vin", "VIN123"); + row.put("phone", "013800000000"); + return row; + } + + private static Map locationResult(String factId, String ts) { + Map row = new LinkedHashMap<>(); + row.put("ts", Timestamp.from(Instant.parse(ts))); + row.put("fact_id", factId); + row.put("frame_id", "frame-1"); + row.put("received_at", Timestamp.from(Instant.parse(ts).plusMillis(500))); + row.put("longitude", 113.12); + row.put("latitude", 23.45); + row.put("altitude_m", 8.0); + row.put("speed_kmh", 42.5); + row.put("direction_deg", 90.0); + row.put("alarm_flag", 1L); + row.put("status_flag", 3L); + row.put("total_mileage_km", null); + row.put("raw_uri", "archive://gb32960/frame-1.bin"); + row.put("metadata_json", "{}"); + row.put("protocol", "GB32960"); + row.put("vehicle_key", "vin:VIN123"); + row.put("vin", "VIN123"); + row.put("phone", ""); + return row; + } + + private static final class RecordingQueryJdbc { + private final List> rows = new ArrayList<>(); + private final Map boundValues = new HashMap<>(); + private String preparedSql; + + DataSource dataSource() { + return (DataSource) Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[]{DataSource.class}, + (proxy, method, args) -> switch (method.getName()) { + case "getConnection" -> connection(); + default -> null; + }); + } + + private Connection connection() { + return (Connection) Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[]{Connection.class}, + (proxy, method, args) -> switch (method.getName()) { + case "prepareStatement" -> preparedStatement((String) args[0]); + case "getAutoCommit" -> true; + default -> null; + }); + } + + private PreparedStatement preparedStatement(String sql) { + preparedSql = sql; + return (PreparedStatement) Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[]{PreparedStatement.class}, + (proxy, method, args) -> switch (method.getName()) { + case "setTimestamp", "setObject" -> { + boundValues.put((Integer) args[0], args[1]); + yield null; + } + case "executeQuery" -> resultSet(); + default -> null; + }); + } + + private ResultSet resultSet() { + final int[] index = {-1}; + return (ResultSet) Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[]{ResultSet.class}, + (proxy, method, args) -> switch (method.getName()) { + case "next" -> ++index[0] < rows.size(); + case "getTimestamp", "getString", "getInt", "getLong", "getDouble", "getObject" -> + rows.get(index[0]).get((String) args[0]); + case "wasNull" -> false; + default -> null; + }); + } + } +} diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java index b832c264..58d8ba0e 100644 --- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java @@ -1,6 +1,7 @@ package com.lingniu.ingest.tdenginehistory.config; import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema; +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; @@ -44,4 +45,12 @@ class TdengineHistoryAutoConfigurationTest { .withPropertyValues("lingniu.ingest.tdengine-history.enabled=true") .run(context -> assertThat(context).hasSingleBean(TdengineHistoryWriter.class)); } + + @Test + void createsReaderWhenEnabledAndDataSourceExists() { + contextRunner + .withBean(DataSource.class, () -> mock(DataSource.class)) + .withPropertyValues("lingniu.ingest.tdengine-history.enabled=true") + .run(context -> assertThat(context).hasSingleBean(TdengineHistoryReader.class)); + } }