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 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, 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); ShadowReadCursor queryCursor = cursor.isInitial() ? new ShadowReadCursor(databaseStartedAt, "") : cursor; queryCursor = queryCursor.withOverlap(CURSOR_OVERLAP); List queues = immutable(source.readQueues( queryCursor, deadline)); requireLimit("QUEUE", queues.size(), QUEUE_LIMIT); List vehicles = immutable(source.readVehicles(deadline)); requireLimit("VEHICLE", vehicles.size(), VEHICLE_LIMIT); List 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 List immutable(List values) { if (values == null) { throw new ShadowFactReadException("SOURCE_RETURNED_NULL_FACT_LIST", Duration.ZERO); } return Collections.unmodifiableList(new ArrayList(values)); } public interface FactSource { Instant readDatabaseClock(Instant deadline); List readQueues(ShadowReadCursor cursor, Instant deadline); List readVehicles(Instant deadline); List 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 " + "WHERE NULLIF(LTRIM(RTRIM(CONVERT(nvarchar(256), site_id))), N'') " + "IS NOT NULL 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) 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 readQueues(ShadowReadCursor cursor, Instant deadline) { List 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 readVehicles(Instant deadline) { return jdbc.execute((ConnectionCallback>) connection -> { try (PreparedStatement statement = connection.prepareStatement( VEHICLE_SQL)) { statement.setInt(1, VEHICLE_LIMIT + 1); timeout(statement, deadline); try (ResultSet result = statement.executeQuery()) { List facts = new ArrayList(); 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 readMaps(Instant deadline) { return jdbc.execute((ConnectionCallback>) connection -> { try (PreparedStatement statement = connection.prepareStatement( MAP_SQL)) { statement.setInt(1, MAP_LIMIT + 1); timeout(statement, deadline); try (ResultSet result = statement.executeQuery()) { List facts = new ArrayList(); 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 queryQueues(String table, ShadowReadCursor cursor, int limit, Instant deadline, String suffix) { String sql = QUEUE_SQL_PREFIX + "FROM " + table + suffix; return jdbc.execute((ConnectionCallback>) 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 facts = new ArrayList(); 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; } } }