feat(javaTest): expose local dispatch visualizer API

2026-08-05 15:16:52 +08:00
parent a0453d8b22
commit f80f92dca3
5 changed files with 488 additions and 0 deletions
--- a/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerApplication.java
+++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerApplication.java
@@ -0,0 +1,38 @@
+package com.ft546.wcs.shadow.history.visualizer;
+
+import com.ft546.wcs.shadow.history.HistoricalQueueDataset;
+import com.ft546.wcs.shadow.history.HistoricalQueueLoader;
+import com.ft546.wcs.shadow.history.simulation.HistoricalSimulationInput;
+import com.ft546.wcs.shadow.history.simulation.HistoricalSimulationInputFactory;
+
+import java.nio.file.Paths;
+import java.util.concurrent.CountDownLatch;
+
+/** 启动只监听本机的调度过程可视化审计台。 */
+public final class HistoricalDispatchVisualizerApplication {
+    private HistoricalDispatchVisualizerApplication() { }
+
+    public static void main(String[] args) throws Exception {
+        if (args.length > 1) {
+            throw new IllegalArgumentException(
+                    "usage: HistoricalDispatchVisualizerApplication [port]");
+        }
+        int port = args.length == 0 ? 18084 : Integer.parseInt(args[0]);
+        ClassLoader classLoader = HistoricalDispatchVisualizerApplication.class.getClassLoader();
+        HistoricalQueueLoader loader = new HistoricalQueueLoader();
+        HistoricalQueueDataset dataset = classLoader.getResource("shadow/history/manifest.json") != null
+                ? loader.load(classLoader, "shadow/history")
+                : loader.load(Paths.get("src/test/resources/shadow/history"));
+        HistoricalSimulationInput input = new HistoricalSimulationInputFactory().create(dataset);
+        HistoricalDispatchVisualizerServer server =
+                new HistoricalDispatchVisualizerServer(port, dataset, input);
+        CountDownLatch shutdown = new CountDownLatch(1);
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            server.stop();
+            shutdown.countDown();
+        }, "javaTest-visualizer-shutdown"));
+        server.start();
+        System.out.println("DISPATCH-VISUALIZER http://127.0.0.1:" + server.getPort());
+        shutdown.await();
+    }
+}
--- a/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerServer.java
+++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerServer.java
@@ -0,0 +1,235 @@
+package com.ft546.wcs.shadow.history.visualizer;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.ft546.wcs.shadow.history.HistoricalQueueDataset;
+import com.ft546.wcs.shadow.history.simulation.HistoricalDispatchSimulationResult;
+import com.ft546.wcs.shadow.history.simulation.HistoricalSimulationInput;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * 为离线调度审计页面提供同源静态资源和 JSON API。
+ *
+ * <p>服务只绑定到本机回环地址，不暴露生产系统接口，也不接受文件路径或命令输入。
+ */
+public final class HistoricalDispatchVisualizerServer {
+    private static final int MAX_REQUEST_BYTES = 64 * 1024;
+
+    private final HttpServer httpServer;
+    private final ExecutorService executor;
+    private final ObjectMapper objectMapper;
+    private final HistoricalSimulationInput input;
+    private final PointToPointSimulationService pointService;
+    private final HistoricalWorkloadSimulationService historicalService;
+    private final SimulationVisualizationMapper mapper;
+    private final AtomicBoolean simulationBusy = new AtomicBoolean(false);
+    private final AtomicBoolean started = new AtomicBoolean(false);
+
+    public HistoricalDispatchVisualizerServer(
+            int port, HistoricalQueueDataset dataset, HistoricalSimulationInput input)
+            throws IOException {
+        if (port < 0 || port > 65_535) {
+            throw new IllegalArgumentException("port must be between 0 and 65535");
+        }
+        if (dataset == null || input == null) {
+            throw new IllegalArgumentException("dataset and input are required");
+        }
+        this.input = input;
+        this.pointService = new PointToPointSimulationService(input);
+        this.historicalService = new HistoricalWorkloadSimulationService(input);
+        this.mapper = new SimulationVisualizationMapper(dataset);
+        this.objectMapper = new ObjectMapper();
+        this.httpServer = HttpServer.create(new InetSocketAddress(
+                InetAddress.getByName("127.0.0.1"), port), 0);
+        this.executor = Executors.newFixedThreadPool(4, new VisualizerThreadFactory());
+        this.httpServer.setExecutor(executor);
+        this.httpServer.createContext("/", new DispatchHandler());
+    }
+
+    /** 启动本地服务。重复启动属于调用错误。 */
+    public void start() {
+        if (!started.compareAndSet(false, true)) {
+            throw new IllegalStateException("visualizer server is already started");
+        }
+        httpServer.start();
+    }
+
+    /** 停止服务并释放测试或命令行进程使用的线程。 */
+    public void stop() {
+        if (started.compareAndSet(true, false)) httpServer.stop(0);
+        executor.shutdownNow();
+    }
+
+    public int getPort() {
+        return httpServer.getAddress().getPort();
+    }
+
+    private final class DispatchHandler implements HttpHandler {
+        @Override
+        public void handle(HttpExchange exchange) throws IOException {
+            try {
+                route(exchange);
+            } catch (VisualizerHttpException exception) {
+                writeError(exchange, exception.getStatus(), exception.getCode(),
+                        exception.getMessage());
+            } catch (JsonProcessingException exception) {
+                writeError(exchange, 400, "BAD_REQUEST", "请求 JSON 无法解析");
+            } catch (IllegalArgumentException exception) {
+                writeError(exchange, 400, businessCode(exception.getMessage()),
+                        exception.getMessage());
+            } catch (IllegalStateException exception) {
+                String code = exception.getMessage() != null
+                        && exception.getMessage().contains("tick limit")
+                        ? "TICK_LIMIT_EXCEEDED" : "WATCHDOG_DEADLOCK";
+                writeError(exchange, 422, code, exception.getMessage());
+            } catch (Exception exception) {
+                writeError(exchange, 500, "INTERNAL_ERROR", "调度审计服务处理失败");
+            } finally {
+                exchange.close();
+            }
+        }
+    }
+
+    private void route(HttpExchange exchange) throws IOException {
+        String path = exchange.getRequestURI().getPath();
+        String method = exchange.getRequestMethod();
+        if ("/api/health".equals(path)) {
+            requireMethod(method, "GET");
+            Map<String, Object> health = new LinkedHashMap<String, Object>();
+            health.put("status", "UP");
+            health.put("mode", "OFFLINE_IN_MEMORY_SIMULATION");
+            health.put("mapLoaded", !input.getGridMap().getSites().isEmpty());
+            health.put("simulationBusy", simulationBusy.get());
+            writeJson(exchange, 200, health);
+            return;
+        }
+        if ("/api/bootstrap".equals(path)) {
+            requireMethod(method, "GET");
+            writeJson(exchange, 200, mapper.bootstrap(input));
+            return;
+        }
+        if ("/api/simulations/point-to-point".equals(path)) {
+            requireMethod(method, "POST");
+            byte[] body = readBody(exchange);
+            PointToPointSimulationRequest request = objectMapper.readValue(
+                    body, PointToPointSimulationRequest.class);
+            writeJson(exchange, 200, runSimulation(() ->
+                    mapper.result(input, pointService.run(request))));
+            return;
+        }
+        if ("/api/simulations/historical".equals(path)) {
+            requireMethod(method, "POST");
+            readBody(exchange);
+            writeJson(exchange, 200, runSimulation(() -> {
+                HistoricalDispatchSimulationResult result = historicalService.run();
+                return mapper.result(input, result);
+            }));
+            return;
+        }
+        throw new VisualizerHttpException(404, "NOT_FOUND", "请求的审计资源不存在");
+    }
+
+    private Map<String, Object> runSimulation(SimulationCall call) {
+        if (!simulationBusy.compareAndSet(false, true)) {
+            throw new VisualizerHttpException(409, "SIMULATION_BUSY", "已有仿真正在运行");
+        }
+        try {
+            return call.run();
+        } finally {
+            simulationBusy.set(false);
+        }
+    }
+
+    private byte[] readBody(HttpExchange exchange) throws IOException {
+        String contentLength = exchange.getRequestHeaders().getFirst("Content-Length");
+        if (contentLength != null) {
+            try {
+                if (Long.parseLong(contentLength) > MAX_REQUEST_BYTES) {
+                    throw new VisualizerHttpException(413, "PAYLOAD_TOO_LARGE", "请求体超过 64 KiB");
+                }
+            } catch (NumberFormatException exception) {
+                throw new VisualizerHttpException(400, "BAD_REQUEST", "Content-Length 无效");
+            }
+        }
+        try (InputStream inputStream = exchange.getRequestBody();
+             ByteArrayOutputStream output = new ByteArrayOutputStream()) {
+            byte[] buffer = new byte[4096];
+            int read;
+            while ((read = inputStream.read(buffer)) >= 0) {
+                if (output.size() + read > MAX_REQUEST_BYTES) {
+                    throw new VisualizerHttpException(413, "PAYLOAD_TOO_LARGE", "请求体超过 64 KiB");
+                }
+                output.write(buffer, 0, read);
+            }
+            return output.toByteArray();
+        }
+    }
+
+    private void requireMethod(String actual, String expected) {
+        if (!expected.equals(actual)) {
+            throw new VisualizerHttpException(405, "METHOD_NOT_ALLOWED",
+                    "该接口仅允许 " + expected);
+        }
+    }
+
+    private void writeError(HttpExchange exchange, int status, String code, String message)
+            throws IOException {
+        Map<String, Object> value = new LinkedHashMap<String, Object>();
+        value.put("code", code);
+        value.put("message", message == null ? "请求处理失败" : message);
+        value.put("details", new LinkedHashMap<String, Object>());
+        writeJson(exchange, status, value);
+    }
+
+    private void writeJson(HttpExchange exchange, int status, Object value) throws IOException {
+        byte[] body = objectMapper.writeValueAsBytes(value);
+        exchange.getResponseHeaders().set("Content-Type", "application/json; charset=utf-8");
+        exchange.getResponseHeaders().set("Cache-Control", "no-store");
+        exchange.getResponseHeaders().set("X-Content-Type-Options", "nosniff");
+        exchange.sendResponseHeaders(status, body.length);
+        try (OutputStream output = exchange.getResponseBody()) {
+            output.write(body);
+        }
+    }
+
+    private String businessCode(String message) {
+        if (message == null) return "BAD_REQUEST";
+        int separator = message.indexOf(':');
+        if (separator <= 0) return "BAD_REQUEST";
+        String candidate = message.substring(0, separator);
+        return candidate.matches("[A-Z_]+") ? candidate : "BAD_REQUEST";
+    }
+
+    private interface SimulationCall {
+        Map<String, Object> run();
+    }
+
+    private static final class VisualizerThreadFactory implements ThreadFactory {
+        private final AtomicInteger sequence = new AtomicInteger();
+
+        @Override
+        public Thread newThread(Runnable runnable) {
+            Thread thread = new Thread(runnable,
+                    "javaTest-visualizer-http-" + sequence.incrementAndGet());
+            thread.setDaemon(true);
+            return thread;
+        }
+    }
+}
--- a/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalWorkloadSimulationService.java
+++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalWorkloadSimulationService.java
@@ -0,0 +1,28 @@
+package com.ft546.wcs.shadow.history.visualizer;
+
+import com.ft546.wcs.shadow.history.simulation.DeterministicFaultPlan;
+import com.ft546.wcs.shadow.history.simulation.HistoricalDispatchSimulationResult;
+import com.ft546.wcs.shadow.history.simulation.HistoricalDispatchSimulator;
+import com.ft546.wcs.shadow.history.simulation.HistoricalSimulationInput;
+import com.ft546.wcs.shadow.history.simulation.SimulatedPlcGateway;
+import com.ft546.wcs.shadow.history.simulation.SimulationConfig;
+
+/** 重新运行固化的 240 条历史负载，并保留既定的确定性 PLC 故障注入。 */
+public final class HistoricalWorkloadSimulationService {
+    private final HistoricalSimulationInput input;
+
+    public HistoricalWorkloadSimulationService(HistoricalSimulationInput input) {
+        if (input == null) throw new IllegalArgumentException("input is required");
+        this.input = input;
+    }
+
+    public HistoricalDispatchSimulationResult run() {
+        DeterministicFaultPlan.Builder faults = DeterministicFaultPlan.builder();
+        for (int i = 0; i < input.getTasks().size(); i += 37) {
+            faults.plcFailure(input.getTasks().get(i).getSimulationTaskId(), 1,
+                    SimulatedPlcGateway.PlcReason.SEND_FAILURE);
+        }
+        return new HistoricalDispatchSimulator().simulate(
+                input, SimulationConfig.defaults(), faults.build());
+    }
+}
--- a/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/VisualizerHttpException.java
+++ b/javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/VisualizerHttpException.java
@@ -0,0 +1,16 @@
+package com.ft546.wcs.shadow.history.visualizer;
+
+/** 表示可稳定映射为前端业务错误码的 HTTP 请求异常。 */
+final class VisualizerHttpException extends RuntimeException {
+    private final int status;
+    private final String code;
+
+    VisualizerHttpException(int status, String code, String message) {
+        super(message);
+        this.status = status;
+        this.code = code;
+    }
+
+    int getStatus() { return status; }
+    String getCode() { return code; }
+}