feat(ft546): collect bounded read-only shadow facts
This commit is contained in:
@@ -0,0 +1,388 @@
|
||||
package com.ruoyi.shadow.facts;
|
||||
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.jdbc.core.ConnectionCallback;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Timestamp;
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Calendar;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.TimeZone;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public final class SqlServerShadowFactReader {
|
||||
static final int QUEUE_LIMIT = 500;
|
||||
static final int VEHICLE_LIMIT = 100;
|
||||
static final int MAP_LIMIT = 10_000;
|
||||
private static final Duration CURSOR_OVERLAP = Duration.ofSeconds(5);
|
||||
private static final Duration MAXIMUM_BATCH_DURATION = Duration.ofSeconds(4);
|
||||
private static final Duration[] BACKOFF = {
|
||||
Duration.ofSeconds(1), Duration.ofSeconds(2),
|
||||
Duration.ofSeconds(4), Duration.ofSeconds(8),
|
||||
Duration.ofSeconds(30)
|
||||
};
|
||||
|
||||
private final FactSource source;
|
||||
private final TransactionTemplate sourceTransaction;
|
||||
private final Clock clock;
|
||||
private final AtomicBoolean inFlight = new AtomicBoolean();
|
||||
private final AtomicInteger consecutiveFailures = new AtomicInteger();
|
||||
private volatile List<ShadowMapFact> cachedMaps;
|
||||
|
||||
public SqlServerShadowFactReader(FactSource source,
|
||||
TransactionTemplate sourceTransaction,
|
||||
Clock clock) {
|
||||
if (source == null || sourceTransaction == null || clock == null) {
|
||||
throw new IllegalArgumentException(
|
||||
"source, sourceTransaction, and clock are required");
|
||||
}
|
||||
this.source = source;
|
||||
this.sourceTransaction = sourceTransaction;
|
||||
this.sourceTransaction.setReadOnly(true);
|
||||
this.clock = clock;
|
||||
}
|
||||
|
||||
public ShadowFactBatch read(ShadowReadCursor cursor, Instant deadline) {
|
||||
if (cursor == null || deadline == null) {
|
||||
throw new IllegalArgumentException("cursor and deadline are required");
|
||||
}
|
||||
Instant startedAt = clock.instant();
|
||||
Duration requested = Duration.between(startedAt, deadline);
|
||||
if (requested.isNegative() || requested.isZero()) {
|
||||
throw immediate("SOURCE_BATCH_DEADLINE_EXCEEDED");
|
||||
}
|
||||
if (requested.compareTo(MAXIMUM_BATCH_DURATION) > 0) {
|
||||
throw immediate("SOURCE_BATCH_DEADLINE_MUST_NOT_EXCEED_4_SECONDS");
|
||||
}
|
||||
if (!inFlight.compareAndSet(false, true)) {
|
||||
throw immediate("SOURCE_CYCLE_ALREADY_IN_FLIGHT");
|
||||
}
|
||||
try {
|
||||
ShadowFactBatch batch = sourceTransaction.execute(status ->
|
||||
collect(cursor.withOverlap(CURSOR_OVERLAP), deadline));
|
||||
if (batch == null) {
|
||||
throw new ShadowFactReadException("SOURCE_EMPTY_BATCH",
|
||||
Duration.ZERO);
|
||||
}
|
||||
consecutiveFailures.set(0);
|
||||
return batch;
|
||||
} catch (ShadowFactReadException failure) {
|
||||
if (failure.getReasonCode().startsWith("SOURCE_CYCLE_")
|
||||
|| failure.getReasonCode().contains("DEADLINE_MUST_NOT")) {
|
||||
throw failure;
|
||||
}
|
||||
throw failed(failure.getReasonCode(), failure);
|
||||
} catch (DataAccessException failure) {
|
||||
throw failed(classify(failure), failure);
|
||||
} catch (RuntimeException failure) {
|
||||
throw failed("SOURCE_READ_FAILED", failure);
|
||||
} finally {
|
||||
inFlight.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
private ShadowFactBatch collect(ShadowReadCursor cursor, Instant deadline) {
|
||||
ensureTime(deadline);
|
||||
Instant databaseStartedAt = source.readDatabaseClock(deadline);
|
||||
List<ShadowQueueFact> queues = immutable(source.readQueues(cursor, deadline));
|
||||
requireLimit("QUEUE", queues.size(), QUEUE_LIMIT);
|
||||
List<ShadowVehicleFact> vehicles =
|
||||
immutable(source.readVehicles(deadline));
|
||||
requireLimit("VEHICLE", vehicles.size(), VEHICLE_LIMIT);
|
||||
List<ShadowMapFact> maps = cachedMaps;
|
||||
if (maps == null) {
|
||||
maps = immutable(source.readMaps(deadline));
|
||||
requireLimit("MAP", maps.size(), MAP_LIMIT);
|
||||
}
|
||||
ensureTime(deadline);
|
||||
Instant databaseCompletedAt = source.readDatabaseClock(deadline);
|
||||
ensureTime(deadline);
|
||||
ShadowFactBatch batch = ShadowFactBatch.create(databaseStartedAt,
|
||||
databaseCompletedAt, queues, vehicles, maps);
|
||||
if (cachedMaps == null) {
|
||||
cachedMaps = maps;
|
||||
}
|
||||
return batch;
|
||||
}
|
||||
|
||||
private void ensureTime(Instant deadline) {
|
||||
queryTimeoutSeconds(clock.instant(), deadline);
|
||||
}
|
||||
|
||||
public static int queryTimeoutSeconds(Instant now, Instant deadline) {
|
||||
if (now == null || deadline == null || !deadline.isAfter(now)) {
|
||||
throw immediate("SOURCE_BATCH_DEADLINE_EXCEEDED");
|
||||
}
|
||||
long remainingMillis = Duration.between(now, deadline).toMillis();
|
||||
long completeSeconds = remainingMillis / 1_000L;
|
||||
if (completeSeconds < 1L) {
|
||||
throw immediate("SOURCE_BATCH_DEADLINE_EXCEEDED");
|
||||
}
|
||||
return (int) Math.min(3L, completeSeconds);
|
||||
}
|
||||
|
||||
private static void requireLimit(String kind, int actual, int limit) {
|
||||
if (actual > limit) {
|
||||
throw new ShadowFactReadException(
|
||||
kind + "_ROW_LIMIT_EXCEEDED:" + limit, Duration.ZERO);
|
||||
}
|
||||
}
|
||||
|
||||
private ShadowFactReadException failed(String reason, Throwable cause) {
|
||||
int failure = consecutiveFailures.getAndIncrement();
|
||||
Duration delay = BACKOFF[Math.min(failure, BACKOFF.length - 1)];
|
||||
return new ShadowFactReadException(reason, delay, cause);
|
||||
}
|
||||
|
||||
private static ShadowFactReadException immediate(String reason) {
|
||||
return new ShadowFactReadException(reason, Duration.ZERO);
|
||||
}
|
||||
|
||||
private static String classify(DataAccessException failure) {
|
||||
String name = failure.getClass().getSimpleName().toUpperCase();
|
||||
if (name.contains("TIMEOUT") || name.contains("TRANSIENT")) {
|
||||
return "SOURCE_QUERY_TIMEOUT";
|
||||
}
|
||||
return "SOURCE_UNAVAILABLE";
|
||||
}
|
||||
|
||||
private static <T> List<T> immutable(List<T> values) {
|
||||
if (values == null) {
|
||||
throw new ShadowFactReadException("SOURCE_RETURNED_NULL_FACT_LIST",
|
||||
Duration.ZERO);
|
||||
}
|
||||
return Collections.unmodifiableList(new ArrayList<T>(values));
|
||||
}
|
||||
|
||||
public interface FactSource {
|
||||
Instant readDatabaseClock(Instant deadline);
|
||||
List<ShadowQueueFact> readQueues(ShadowReadCursor cursor,
|
||||
Instant deadline);
|
||||
List<ShadowVehicleFact> readVehicles(Instant deadline);
|
||||
List<ShadowMapFact> readMaps(Instant deadline);
|
||||
}
|
||||
|
||||
public static final class JdbcFactSource implements FactSource {
|
||||
static final String CLOCK_SQL =
|
||||
"SELECT database_utc_time FROM shadow_ro.v_current_database_clock";
|
||||
static final String QUEUE_SQL_PREFIX =
|
||||
"SELECT TOP (?) guid, task_no, task_type, priority, "
|
||||
+ "cargo_pick_site, cargo_place_site, start_site, end_site, "
|
||||
+ "create_date_utc, agv_id, path_info, trun_path, default02, "
|
||||
+ "default03, default04, default05, default06, default07, default08 ";
|
||||
static final String QUEUE_SQL_SUFFIX =
|
||||
" WHERE create_date_utc > ? "
|
||||
+ "OR (create_date_utc = ? AND guid > ?) "
|
||||
+ "ORDER BY create_date_utc, guid";
|
||||
static final String ACTIVE_QUEUE_SQL_SUFFIX =
|
||||
" ORDER BY create_date_utc, guid";
|
||||
static final String VEHICLE_SQL =
|
||||
"SELECT TOP (?) guid, device_id, agv_status, task_no, src_site, "
|
||||
+ "tar_site, curr_site, task_type, battery, work_mode, alarm, "
|
||||
+ "status, finish, charge_flag, work_phase, err_msg, "
|
||||
+ "change_layer, cargo_flag, path, allow_send, curr_floor, "
|
||||
+ "locked, spare2, spare5, last_time_utc "
|
||||
+ "FROM shadow_ro.v_device_srgv ORDER BY device_id, guid";
|
||||
static final String MAP_SQL =
|
||||
"SELECT TOP (?) guid, site_id, x, y, z, value "
|
||||
+ "FROM shadow_ro.v_wcs_srgv_map ORDER BY site_id, guid";
|
||||
|
||||
private final JdbcTemplate jdbc;
|
||||
private final Clock clock;
|
||||
|
||||
public JdbcFactSource(JdbcTemplate jdbc, Clock clock) {
|
||||
if (jdbc == null || clock == null) {
|
||||
throw new IllegalArgumentException("jdbc and clock are required");
|
||||
}
|
||||
this.jdbc = jdbc;
|
||||
this.clock = clock;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Instant readDatabaseClock(Instant deadline) {
|
||||
return jdbc.execute((ConnectionCallback<Instant>) connection -> {
|
||||
try (PreparedStatement statement = connection.prepareStatement(
|
||||
CLOCK_SQL)) {
|
||||
timeout(statement, deadline);
|
||||
try (ResultSet result = statement.executeQuery()) {
|
||||
if (!result.next()) {
|
||||
throw new ShadowFactReadException(
|
||||
"SOURCE_DATABASE_CLOCK_MISSING", Duration.ZERO);
|
||||
}
|
||||
return instant(result, "database_utc_time");
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ShadowQueueFact> readQueues(ShadowReadCursor cursor,
|
||||
Instant deadline) {
|
||||
List<ShadowQueueFact> result = queryQueues(
|
||||
"shadow_ro.v_wcs_srgv_queue", null, QUEUE_LIMIT + 1,
|
||||
deadline, ACTIVE_QUEUE_SQL_SUFFIX);
|
||||
if (result.size() > QUEUE_LIMIT) {
|
||||
return result;
|
||||
}
|
||||
int remainingWithOverflow = QUEUE_LIMIT - result.size() + 1;
|
||||
result.addAll(queryQueues("shadow_ro.v_wcs_srgv_queue_finish",
|
||||
cursor, remainingWithOverflow, deadline, QUEUE_SQL_SUFFIX));
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ShadowVehicleFact> readVehicles(Instant deadline) {
|
||||
return jdbc.execute((ConnectionCallback<List<ShadowVehicleFact>>)
|
||||
connection -> {
|
||||
try (PreparedStatement statement = connection.prepareStatement(
|
||||
VEHICLE_SQL)) {
|
||||
statement.setInt(1, VEHICLE_LIMIT + 1);
|
||||
timeout(statement, deadline);
|
||||
try (ResultSet result = statement.executeQuery()) {
|
||||
List<ShadowVehicleFact> facts =
|
||||
new ArrayList<ShadowVehicleFact>();
|
||||
while (result.next()) {
|
||||
facts.add(ShadowVehicleFact.builder(
|
||||
result.getString("guid"),
|
||||
result.getString("device_id"))
|
||||
.agvStatus(result.getString("agv_status"))
|
||||
.taskNo(intValue(result, "task_no"))
|
||||
.sourceSite(result.getString("src_site"))
|
||||
.targetSite(result.getString("tar_site"))
|
||||
.currentSite(result.getString("curr_site"))
|
||||
.taskType(intValue(result, "task_type"))
|
||||
.battery(intValue(result, "battery"))
|
||||
.workMode(intValue(result, "work_mode"))
|
||||
.alarm(intValue(result, "alarm"))
|
||||
.status(intValue(result, "status"))
|
||||
.finish(intValue(result, "finish"))
|
||||
.chargeFlag(intValue(result, "charge_flag"))
|
||||
.workPhase(intValue(result, "work_phase"))
|
||||
.errorCode(intValue(result, "err_msg"))
|
||||
.changeLayer(intValue(result, "change_layer"))
|
||||
.cargoFlag(intValue(result, "cargo_flag"))
|
||||
.path(result.getString("path"))
|
||||
.allowSend(intValue(result, "allow_send"))
|
||||
.currentFloor(intValue(result, "curr_floor"))
|
||||
.locked(intValue(result, "locked"))
|
||||
.spare2(result.getString("spare2"))
|
||||
.spare5(result.getString("spare5"))
|
||||
.capturedAt(instant(result, "last_time_utc"))
|
||||
.build());
|
||||
}
|
||||
return facts;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ShadowMapFact> readMaps(Instant deadline) {
|
||||
return jdbc.execute((ConnectionCallback<List<ShadowMapFact>>)
|
||||
connection -> {
|
||||
try (PreparedStatement statement = connection.prepareStatement(
|
||||
MAP_SQL)) {
|
||||
statement.setInt(1, MAP_LIMIT + 1);
|
||||
timeout(statement, deadline);
|
||||
try (ResultSet result = statement.executeQuery()) {
|
||||
List<ShadowMapFact> facts = new ArrayList<ShadowMapFact>();
|
||||
while (result.next()) {
|
||||
facts.add(new ShadowMapFact(result.getString("guid"),
|
||||
result.getString("site_id"),
|
||||
result.getInt("x"), result.getInt("y"),
|
||||
result.getInt("z"), result.getInt("value")));
|
||||
}
|
||||
return facts;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private List<ShadowQueueFact> queryQueues(String table,
|
||||
ShadowReadCursor cursor,
|
||||
int limit,
|
||||
Instant deadline,
|
||||
String suffix) {
|
||||
String sql = QUEUE_SQL_PREFIX + "FROM " + table + suffix;
|
||||
return jdbc.execute((ConnectionCallback<List<ShadowQueueFact>>)
|
||||
connection -> {
|
||||
try (PreparedStatement statement = connection.prepareStatement(sql)) {
|
||||
statement.setInt(1, limit);
|
||||
if (cursor != null) {
|
||||
Timestamp timestamp = Timestamp.from(
|
||||
cursor.getLastCreateDate());
|
||||
statement.setTimestamp(2, timestamp, utcCalendar());
|
||||
statement.setTimestamp(3, timestamp, utcCalendar());
|
||||
statement.setString(4, cursor.getLastGuid());
|
||||
}
|
||||
timeout(statement, deadline);
|
||||
try (ResultSet result = statement.executeQuery()) {
|
||||
List<ShadowQueueFact> facts =
|
||||
new ArrayList<ShadowQueueFact>();
|
||||
while (result.next()) {
|
||||
facts.add(ShadowQueueFact.builder(table,
|
||||
result.getString("guid"))
|
||||
.taskNo(result.getString("task_no"))
|
||||
.taskType(result.getString("task_type"))
|
||||
.priority(longValue(result, "priority"))
|
||||
.cargoPickSite(result.getString("cargo_pick_site"))
|
||||
.cargoPlaceSite(result.getString("cargo_place_site"))
|
||||
.startSite(result.getString("start_site"))
|
||||
.endSite(result.getString("end_site"))
|
||||
.createdAt(instant(result, "create_date_utc"))
|
||||
.agvId(result.getString("agv_id"))
|
||||
.pathInfo(result.getString("path_info"))
|
||||
.turnPath(result.getString("trun_path"))
|
||||
.default02(result.getString("default02"))
|
||||
.default03(result.getString("default03"))
|
||||
.default04(result.getString("default04"))
|
||||
.default05(result.getString("default05"))
|
||||
.default06(result.getString("default06"))
|
||||
.default07(result.getString("default07"))
|
||||
.default08(result.getString("default08"))
|
||||
.build());
|
||||
}
|
||||
return facts;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void timeout(PreparedStatement statement, Instant deadline)
|
||||
throws SQLException {
|
||||
statement.setQueryTimeout(queryTimeoutSeconds(clock.instant(), deadline));
|
||||
}
|
||||
|
||||
private static Instant instant(ResultSet result, String column)
|
||||
throws SQLException {
|
||||
Timestamp timestamp = result.getTimestamp(column, utcCalendar());
|
||||
return timestamp == null ? null : timestamp.toInstant();
|
||||
}
|
||||
|
||||
private static Calendar utcCalendar() {
|
||||
return Calendar.getInstance(TimeZone.getTimeZone("UTC"));
|
||||
}
|
||||
|
||||
private static Long longValue(ResultSet result, String column)
|
||||
throws SQLException {
|
||||
long value = result.getLong(column);
|
||||
return result.wasNull() ? null : value;
|
||||
}
|
||||
|
||||
private static Integer intValue(ResultSet result, String column)
|
||||
throws SQLException {
|
||||
int value = result.getInt(column);
|
||||
return result.wasNull() ? null : value;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user