From f80f92dca3e9b9f2727cb7db2d7bf8e6cd286e04 Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Wed, 5 Aug 2026 15:16:52 +0800 Subject: [PATCH] feat(javaTest): expose local dispatch visualizer API --- ...storicalDispatchVisualizerApplication.java | 38 +++ .../HistoricalDispatchVisualizerServer.java | 235 ++++++++++++++++++ .../HistoricalWorkloadSimulationService.java | 28 +++ .../visualizer/VisualizerHttpException.java | 16 ++ ...istoricalDispatchVisualizerServerTest.java | 171 +++++++++++++ 5 files changed, 488 insertions(+) create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerApplication.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerServer.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/HistoricalWorkloadSimulationService.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/history/visualizer/VisualizerHttpException.java create mode 100644 javaTest/src/test/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerServerTest.java diff --git 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 new file mode 100644 index 0000000..ceab0c8 --- /dev/null +++ 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(); + } +} diff --git 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 new file mode 100644 index 0000000..9b26cee --- /dev/null +++ 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。 + * + *

服务只绑定到本机回环地址，不暴露生产系统接口，也不接受文件路径或命令输入。 + */ +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 health = new LinkedHashMap(); + 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 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 value = new LinkedHashMap(); + value.put("code", code); + value.put("message", message == null ? "请求处理失败" : message); + value.put("details", new LinkedHashMap()); + 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 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; + } + } +} diff --git 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 new file mode 100644 index 0000000..d19b370 --- /dev/null +++ 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()); + } +} diff --git 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 new file mode 100644 index 0000000..cb9c1af --- /dev/null +++ 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; } +} diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerServerTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerServerTest.java new file mode 100644 index 0000000..44530a7 --- /dev/null +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/history/visualizer/HistoricalDispatchVisualizerServerTest.java @@ -0,0 +1,171 @@ +package com.ft546.wcs.shadow.history.visualizer; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +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 com.ft546.wcs.shadow.history.simulation.HistoricalSimulationTask; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class HistoricalDispatchVisualizerServerTest { + private static final ObjectMapper JSON = new ObjectMapper(); + private static HistoricalQueueDataset dataset; + private static HistoricalSimulationInput input; + private static HistoricalSimulationTask sampleTask; + + private HistoricalDispatchVisualizerServer server; + + @BeforeAll + static void loadFixture() { + dataset = new HistoricalQueueLoader().load( + HistoricalDispatchVisualizerServerTest.class.getClassLoader(), "shadow/history"); + input = new HistoricalSimulationInputFactory().create(dataset); + sampleTask = input.getTasks().stream() + .filter(task -> !task.getPhysicalSource().equals(task.getPhysicalTarget())) + .findFirst() + .orElseThrow(() -> new AssertionError("a point-to-point sample is required")); + } + + @BeforeEach + void startServer() throws IOException { + server = new HistoricalDispatchVisualizerServer(0, dataset, input); + server.start(); + } + + @AfterEach + void stopServer() { + if (server != null) server.stop(); + } + + @Test + void servesHealthBootstrapAndPointToPointSimulation() throws Exception { + HttpResult health = request("GET", "/api/health", null); + HttpResult bootstrap = request("GET", "/api/bootstrap", null); + PointToPointSimulationRequest task = PointToPointSimulationRequest.valid( + sampleTask.getPhysicalSource(), sampleTask.getPhysicalTarget(), + "INBOUND", 1, 100, null, Collections.emptyList(), "NONE"); + HttpResult simulation = request("POST", "/api/simulations/point-to-point", + JSON.writeValueAsString(task)); + + assertEquals(200, health.status); + assertEquals(Boolean.TRUE, map(health.body).get("mapLoaded")); + assertEquals(Boolean.FALSE, map(health.body).get("simulationBusy")); + assertEquals(200, bootstrap.status); + assertEquals("OFFLINE_IN_MEMORY_SIMULATION", map(bootstrap.body).get("mode")); + assertEquals(200, simulation.status); + assertEquals(1, map(map(simulation.body).get("summary")).get("completedCount")); + assertFalse(list(map(simulation.body).get("events")).isEmpty()); + assertTrue(simulation.body.contains("plannedPath")); + assertTrue(simulation.body.contains("MOVE")); + } + + @Test + void historicalEndpointRunsAllCurrentTasks() throws Exception { + HttpResult response = request("POST", "/api/simulations/historical", "{}"); + Map summary = map(map(response.body).get("summary")); + + assertEquals(200, response.status); + assertEquals(240, summary.get("inputCount")); + assertEquals(240, summary.get("completedCount")); + assertEquals(0, summary.get("waitingCount")); + assertEquals(0, summary.get("failedCount")); + assertEquals(Boolean.FALSE, map(response.body).get("pathInfoUsedForPlanning")); + assertEquals(0, map(map(response.body).get("invariants")).get("totalSafetyViolations")); + } + + @Test + void rejectsUnknownRoutesWrongMethodsMalformedAndOversizedBodies() throws Exception { + HttpResult unknown = request("GET", "/api/unknown", null); + HttpResult wrongMethod = request("POST", "/api/health", "{}"); + HttpResult malformed = request("POST", "/api/simulations/point-to-point", "{"); + StringBuilder oversized = new StringBuilder(70_000); + for (int i = 0; i < 70_000; i++) oversized.append('x'); + HttpResult tooLarge = request("POST", "/api/simulations/point-to-point", + oversized.toString()); + + assertEquals(404, unknown.status); + assertEquals("NOT_FOUND", map(unknown.body).get("code")); + assertEquals(405, wrongMethod.status); + assertEquals("METHOD_NOT_ALLOWED", map(wrongMethod.body).get("code")); + assertEquals(400, malformed.status); + assertEquals("BAD_REQUEST", map(malformed.body).get("code")); + assertEquals(413, tooLarge.status); + assertEquals("PAYLOAD_TOO_LARGE", map(tooLarge.body).get("code")); + } + + private HttpResult request(String method, String path, String body) throws IOException { + URL url = new URL("http://127.0.0.1:" + server.getPort() + path); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod(method); + connection.setConnectTimeout(5_000); + connection.setReadTimeout(60_000); + connection.setRequestProperty("Accept", "application/json"); + if (body != null) { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + connection.setDoOutput(true); + connection.setRequestProperty("Content-Type", "application/json; charset=utf-8"); + connection.setFixedLengthStreamingMode(bytes.length); + try (OutputStream output = connection.getOutputStream()) { + output.write(bytes); + } + } + int status = connection.getResponseCode(); + InputStream stream = status >= 400 + ? connection.getErrorStream() : connection.getInputStream(); + String response = stream == null ? "" : read(stream); + connection.disconnect(); + return new HttpResult(status, response); + } + + private String read(InputStream input) throws IOException { + try (InputStream source = input; ByteArrayOutputStream output = new ByteArrayOutputStream()) { + byte[] buffer = new byte[4096]; + int read; + while ((read = source.read(buffer)) >= 0) output.write(buffer, 0, read); + return new String(output.toByteArray(), StandardCharsets.UTF_8); + } + } + + private static Map map(String json) throws IOException { + return JSON.readValue(json, new TypeReference>() { }); + } + + @SuppressWarnings("unchecked") + private static Map map(Object value) { + return (Map) value; + } + + @SuppressWarnings("unchecked") + private static java.util.List list(Object value) { + return (java.util.List) value; + } + + private static final class HttpResult { + private final int status; + private final String body; + + private HttpResult(int status, String body) { + this.status = status; + this.body = body; + } + } +}