From c80acf0d994218c997ca1e22413860ea0594bad5 Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Tue, 11 Aug 2026 16:57:55 +0800 Subject: [PATCH] feat(ft546): run standard WCS as the single scheduler --- .../runtime/Ft546AdapterConfiguration.java | 40 ++++ .../ft546/runtime/Ft546WcsScheduler.java | 52 ++++++ .../com/ruoyi/schedule/ScheduleThread.java | 29 +++ .../scheduleTask/srgv/BaseSRgvThread.java | 12 ++ .../runtime/Ft546StandardRuntimeTest.java | 175 ++++++++++++++++++ .../wcs/application/RuntimeTickResult.java | 57 ++++++ .../application/WcsRuntimeCoordinator.java | 96 ++++++++++ .../device/DeviceExecutionService.java | 20 +- .../device/FeedbackReconciler.java | 9 +- .../application/device/SnapshotCollector.java | 6 + .../dispatch/DispatchCoordinator.java | 4 +- .../application/dispatch/GrantedDispatch.java | 20 ++ .../application/recovery/RecoveryGate.java | 6 + .../wcs/domain/dispatch/DispatchTask.java | 29 +++ .../WcsRuntimeCoordinatorTest.java | 125 +++++++++++++ .../device/CommandOutboxDispatcherTest.java | 20 ++ .../device/FeedbackReconcilerTest.java | 18 ++ .../dispatch/DispatchCoordinatorTest.java | 23 +++ 18 files changed, 737 insertions(+), 4 deletions(-) create mode 100644 ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546AdapterConfiguration.java create mode 100644 ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546WcsScheduler.java create mode 100644 ruoyi-admin/src/test/java/com/ruoyi/project/ft546/runtime/Ft546StandardRuntimeTest.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/RuntimeTickResult.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/WcsRuntimeCoordinator.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SnapshotCollector.java create mode 100644 ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/recovery/RecoveryGate.java create mode 100644 ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/WcsRuntimeCoordinatorTest.java diff --git a/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546AdapterConfiguration.java b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546AdapterConfiguration.java new file mode 100644 index 0000000..d36edfe --- /dev/null +++ b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546AdapterConfiguration.java @@ -0,0 +1,40 @@ +package com.ruoyi.project.ft546.runtime; + +import com.ruoyi.wcs.application.WcsRuntimeCoordinator; +import com.ruoyi.wcs.application.device.CommandOutboxDispatcher; +import com.ruoyi.wcs.application.device.DeviceExecutionService; +import com.ruoyi.wcs.application.device.SnapshotCollector; +import com.ruoyi.wcs.application.dispatch.DispatchCoordinator; +import com.ruoyi.wcs.application.recovery.RecoveryGate; +import com.ruoyi.wcs.domain.shared.WcsClock; +import com.ruoyi.wcs.port.out.DispatchSnapshotProvider; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +@ConditionalOnProperty(prefix = "ft546.wcs", name = "runtime-mode", + havingValue = "STANDARD") +public class Ft546AdapterConfiguration { + private static final int OUTBOX_BATCH_LIMIT = 100; + + @Bean + public WcsRuntimeCoordinator ft546WcsRuntimeCoordinator( + SnapshotCollector snapshotCollector, + RecoveryGate recoveryGate, + DispatchSnapshotProvider snapshotProvider, + DispatchCoordinator dispatchCoordinator, + DeviceExecutionService executionService, + CommandOutboxDispatcher outboxDispatcher, + WcsClock clock) { + return new WcsRuntimeCoordinator(snapshotCollector, recoveryGate, + snapshotProvider, dispatchCoordinator, executionService, + outboxDispatcher, clock, OUTBOX_BATCH_LIMIT); + } + + @Bean + public Ft546WcsScheduler ft546WcsScheduler( + WcsRuntimeCoordinator coordinator) { + return new Ft546WcsScheduler(coordinator); + } +} diff --git a/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546WcsScheduler.java b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546WcsScheduler.java new file mode 100644 index 0000000..0cd5661 --- /dev/null +++ b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/runtime/Ft546WcsScheduler.java @@ -0,0 +1,52 @@ +package com.ruoyi.project.ft546.runtime; + +import com.ruoyi.wcs.application.RuntimeTickResult; +import com.ruoyi.wcs.application.WcsRuntimeCoordinator; +import org.springframework.scheduling.annotation.Scheduled; + +import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; + +public final class Ft546WcsScheduler { + private final Supplier tick; + private final AtomicBoolean tickInProgress = new AtomicBoolean(); + private final AtomicLong completedTickCount = new AtomicLong(); + private volatile RuntimeTickResult lastResult; + + public Ft546WcsScheduler(WcsRuntimeCoordinator coordinator) { + this(Objects.requireNonNull(coordinator, "coordinator")::tick); + } + + Ft546WcsScheduler(Supplier tick) { + this.tick = Objects.requireNonNull(tick, "tick"); + } + + @Scheduled(fixedDelayString = + "${ft546.wcs.dispatch-interval-ms:1000}") + public void runScheduledTick() { + if (!tickInProgress.compareAndSet(false, true)) { + return; + } + try { + lastResult = Objects.requireNonNull(tick.get(), + "runtime coordinator returned null"); + completedTickCount.incrementAndGet(); + } finally { + tickInProgress.set(false); + } + } + + public long getCompletedTickCount() { + return completedTickCount.get(); + } + + public RuntimeTickResult getLastResult() { + return lastResult; + } + + public boolean isTickInProgress() { + return tickInProgress.get(); + } +} diff --git a/ruoyi-admin/src/main/java/com/ruoyi/schedule/ScheduleThread.java b/ruoyi-admin/src/main/java/com/ruoyi/schedule/ScheduleThread.java index 96f9ab6..bd9eb50 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/schedule/ScheduleThread.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/schedule/ScheduleThread.java @@ -14,6 +14,7 @@ import com.ruoyi.device.tsj.domain.DeviceTsj; import com.ruoyi.device.tsj.service.IDeviceTsjService; import com.ruoyi.project.ft546.device.Ft546DeviceRuntimeRegistry; import com.ruoyi.project.ft546.device.shuttle.Ft546SrgvRuntimeFactory; +import com.ruoyi.project.ft546.config.Ft546WcsProperties; import com.ruoyi.schedule.plcRead.cdz.PlcCdzRead; import com.ruoyi.schedule.plcRead.ddj.PlcDdjRead; import com.ruoyi.schedule.plcRead.srgv.PlcSrgvRead; @@ -27,6 +28,8 @@ import com.ruoyi.schedule.scheduleTask.ssj.SsjThread; import com.ruoyi.schedule.scheduleTask.tsj.TsjThread; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; @@ -66,6 +69,8 @@ public class ScheduleThread { private Ft546DeviceRuntimeRegistry deviceRuntimeRegistry; @Autowired private Ft546SrgvRuntimeFactory srgvRuntimeFactory; + @Autowired + private Ft546WcsProperties ft546WcsProperties; private volatile boolean isRun = false; private volatile boolean isRun2 = false; @@ -90,10 +95,20 @@ public class ScheduleThread { } + @EventListener(ApplicationReadyEvent.class) + public void startStandardFactReadersAfterApplicationReady() { + if (isStandardMode()) { + startAllThread(); + } + } + /** * 调度初始化 */ private void scheduleInit() { + if (isStandardMode()) { + return; + } MonitoringThread monitoringThread = new MonitoringThread(); monitoringCustomThread = new CustomThread(new Thread(monitoringThread), monitoringThread); SsjThread ssjThread=new SsjThread(); @@ -196,6 +211,10 @@ public class ScheduleThread { * 开启线程2 */ public void startAllThread2() { + if (isStandardMode()) { + throw new IllegalStateException( + "legacy business scheduler is disabled in STANDARD mode"); + } monitoringCustomThread.thread.start(); ssjThreadCustomThread.thread.start(); tsjThreadCustomThread.thread.start(); @@ -322,6 +341,10 @@ public class ScheduleThread { * 停止线程2 */ public void stopAllThread2() { + if (isStandardMode()) { + isRun2 = false; + return; + } monitoringCustomThread.runnable.isRun = false; // ddj1CustomThread.runnable.isRun = false; /*ddj2CustomThread.runnable.isRun = false; @@ -342,4 +365,10 @@ public class ScheduleThread { map.put("isRun2",isRun2); return map; } + + private boolean isStandardMode() { + return ft546WcsProperties != null + && ft546WcsProperties.getRuntimeMode() + == Ft546WcsProperties.RuntimeMode.STANDARD; + } } diff --git a/ruoyi-admin/src/main/java/com/ruoyi/schedule/scheduleTask/srgv/BaseSRgvThread.java b/ruoyi-admin/src/main/java/com/ruoyi/schedule/scheduleTask/srgv/BaseSRgvThread.java index c12dec1..e815b77 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/schedule/scheduleTask/srgv/BaseSRgvThread.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/schedule/scheduleTask/srgv/BaseSRgvThread.java @@ -27,6 +27,7 @@ import com.ruoyi.job.service.impl.JobCallBackServiceImpl; import com.ruoyi.queue.srgv.domain.WcsSrgvQueue; import com.ruoyi.queue.srgv.service.IWcsSrgvQueueFinishService; import com.ruoyi.queue.srgv.service.IWcsSrgvQueueService; +import com.ruoyi.project.ft546.config.Ft546WcsProperties; import com.ruoyi.project.ft546.device.Ft546DeviceRuntimeRegistry; import com.ruoyi.schedule.plcRead.srgv.PlcSrgvRead; import com.ruoyi.schedule.plcRead.srgv.roadUtil.PathFinder; @@ -94,6 +95,10 @@ public class BaseSRgvThread implements Runnable, SRgvThreadInterface { @Override public void run() { + if (isStandardRuntime()) { + log.warn("STANDARD mode owns scheduling; legacy SRGV thread will not run"); + return; + } init(); while (isRun) { @@ -136,6 +141,13 @@ public class BaseSRgvThread implements Runnable, SRgvThreadInterface { } } + private boolean isStandardRuntime() { + Ft546WcsProperties properties = + SpringUtils.getBean(Ft546WcsProperties.class); + return properties.getRuntimeMode() + == Ft546WcsProperties.RuntimeMode.STANDARD; + } + /** * 离开电桩 * diff --git a/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/runtime/Ft546StandardRuntimeTest.java b/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/runtime/Ft546StandardRuntimeTest.java new file mode 100644 index 0000000..b88e2f4 --- /dev/null +++ b/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/runtime/Ft546StandardRuntimeTest.java @@ -0,0 +1,175 @@ +package com.ruoyi.project.ft546.runtime; + +import com.ruoyi.project.ft546.config.Ft546WcsProperties; +import com.ruoyi.common.utils.spring.SpringUtils; +import com.ruoyi.schedule.ScheduleThread; +import com.ruoyi.schedule.scheduleTask.srgv.BaseSRgvThread; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class Ft546StandardRuntimeTest { + + @Test + void standardModeDoesNotConstructLegacyBusinessThreads() { + ScheduleThread runtime = runtime(Ft546WcsProperties.RuntimeMode.STANDARD); + + ReflectionTestUtils.invokeMethod(runtime, "scheduleInit"); + + assertNull(runtime.ssjThreadCustomThread); + assertNull(runtime.sRgvThreadCustomThread); + assertNull(runtime.tsjThreadCustomThread); + assertNull(runtime.cdzThreadCustomThread); + } + + @Test + void legacyModeKeepsExistingBusinessThreadInitialization() { + ScheduleThread runtime = runtime(Ft546WcsProperties.RuntimeMode.LEGACY); + + ReflectionTestUtils.invokeMethod(runtime, "scheduleInit"); + + assertNotNull(runtime.ssjThreadCustomThread); + assertNotNull(runtime.sRgvThreadCustomThread); + assertNotNull(runtime.tsjThreadCustomThread); + assertNotNull(runtime.cdzThreadCustomThread); + } + + @Test + void standardModeRejectsManualLegacySchedulerStart() { + ScheduleThread runtime = runtime(Ft546WcsProperties.RuntimeMode.STANDARD); + + IllegalStateException failure = assertThrows(IllegalStateException.class, + runtime::startAllThread2); + + assertEquals("legacy business scheduler is disabled in STANDARD mode", + failure.getMessage()); + } + + @Test + void baseSrgvThreadRefusesToRunInStandardMode() { + Object originalFactory = ReflectionTestUtils.getField( + SpringUtils.class, "beanFactory"); + DefaultListableBeanFactory factory = + new DefaultListableBeanFactory(); + Ft546WcsProperties properties = new Ft546WcsProperties(); + properties.setRuntimeMode(Ft546WcsProperties.RuntimeMode.STANDARD); + factory.registerSingleton("ft546WcsProperties", properties); + ReflectionTestUtils.setField(SpringUtils.class, "beanFactory", + factory); + AtomicBoolean initialized = new AtomicBoolean(); + BaseSRgvThread legacyThread = new BaseSRgvThread() { + @Override + public void init() { + initialized.set(true); + isRun = false; + } + }; + + try { + legacyThread.run(); + } finally { + ReflectionTestUtils.setField(SpringUtils.class, "beanFactory", + originalFactory); + } + + assertFalse(initialized.get()); + } + + @Test + void runtimeModeDefaultsToLegacy() { + assertEquals(Ft546WcsProperties.RuntimeMode.LEGACY, + new Ft546WcsProperties().getRuntimeMode()); + } + + @Test + void standardSchedulerOwnsTheCoreTick() { + AtomicInteger ticks = new AtomicInteger(); + Ft546WcsScheduler scheduler = new Ft546WcsScheduler(() -> { + ticks.incrementAndGet(); + return com.ruoyi.wcs.application.RuntimeTickResult + .recoveryBlocked(); + }); + + scheduler.runScheduledTick(); + + assertEquals(1, ticks.get()); + assertEquals(1L, scheduler.getCompletedTickCount()); + } + + @Test + void legacyModeDoesNotRequireStandardRuntimePorts() { + new ApplicationContextRunner() + .withUserConfiguration(Ft546AdapterConfiguration.class) + .withPropertyValues("ft546.wcs.runtime-mode=LEGACY") + .run(context -> { + assertNull(context.getStartupFailure()); + assertTrue(context.getBeansOfType( + Ft546WcsScheduler.class).isEmpty()); + }); + } + + @Test + void standardModeFailsFastWhenPersistentRuntimePortsAreMissing() { + new ApplicationContextRunner() + .withUserConfiguration(Ft546AdapterConfiguration.class) + .withPropertyValues("ft546.wcs.runtime-mode=STANDARD") + .run(context -> assertNotNull(context.getStartupFailure())); + } + + @Test + void standardFactReadersStartOnlyAfterApplicationIsReady() { + AtomicInteger starts = new AtomicInteger(); + ScheduleThread runtime = new ScheduleThread() { + @Override + public synchronized void startAllThread() { + starts.incrementAndGet(); + } + }; + Ft546WcsProperties properties = new Ft546WcsProperties(); + properties.setRuntimeMode(Ft546WcsProperties.RuntimeMode.STANDARD); + ReflectionTestUtils.setField(runtime, "ft546WcsProperties", + properties); + + runtime.startStandardFactReadersAfterApplicationReady(); + + assertEquals(1, starts.get()); + } + + @Test + void legacyFactReadersRemainOperatorControlledAfterApplicationReady() { + AtomicInteger starts = new AtomicInteger(); + ScheduleThread runtime = new ScheduleThread() { + @Override + public synchronized void startAllThread() { + starts.incrementAndGet(); + } + }; + ReflectionTestUtils.setField(runtime, "ft546WcsProperties", + new Ft546WcsProperties()); + + runtime.startStandardFactReadersAfterApplicationReady(); + + assertEquals(0, starts.get()); + } + + private static ScheduleThread runtime( + Ft546WcsProperties.RuntimeMode mode) { + Ft546WcsProperties properties = new Ft546WcsProperties(); + properties.setRuntimeMode(mode); + ScheduleThread runtime = new ScheduleThread(); + ReflectionTestUtils.setField(runtime, "ft546WcsProperties", + properties); + return runtime; + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/RuntimeTickResult.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/RuntimeTickResult.java new file mode 100644 index 0000000..48ffd69 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/RuntimeTickResult.java @@ -0,0 +1,57 @@ +package com.ruoyi.wcs.application; + +import com.ruoyi.wcs.application.device.OutboxDispatchResult; +import com.ruoyi.wcs.application.dispatch.DispatchCycleResult; + +public final class RuntimeTickResult { + public enum Status { + COMPLETED, + RECOVERY_BLOCKED + } + + private final Status status; + private final DispatchCycleResult dispatchResult; + private final OutboxDispatchResult outboxResult; + private final int createdCommandCount; + + private RuntimeTickResult(Status status, + DispatchCycleResult dispatchResult, + OutboxDispatchResult outboxResult, + int createdCommandCount) { + this.status = status; + this.dispatchResult = dispatchResult; + this.outboxResult = outboxResult; + this.createdCommandCount = createdCommandCount; + } + + public static RuntimeTickResult completed( + DispatchCycleResult dispatchResult, + OutboxDispatchResult outboxResult, int createdCommandCount) { + if (dispatchResult == null || outboxResult == null + || createdCommandCount < 0) { + throw new IllegalArgumentException("completed tick result is invalid"); + } + return new RuntimeTickResult(Status.COMPLETED, dispatchResult, + outboxResult, createdCommandCount); + } + + public static RuntimeTickResult recoveryBlocked() { + return new RuntimeTickResult(Status.RECOVERY_BLOCKED, null, null, 0); + } + + public Status getStatus() { + return status; + } + + public DispatchCycleResult getDispatchResult() { + return dispatchResult; + } + + public OutboxDispatchResult getOutboxResult() { + return outboxResult; + } + + public int getCreatedCommandCount() { + return createdCommandCount; + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/WcsRuntimeCoordinator.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/WcsRuntimeCoordinator.java new file mode 100644 index 0000000..a9a4312 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/WcsRuntimeCoordinator.java @@ -0,0 +1,96 @@ +package com.ruoyi.wcs.application; + +import com.ruoyi.wcs.application.device.CommandOutboxDispatcher; +import com.ruoyi.wcs.application.device.DeviceExecutionService; +import com.ruoyi.wcs.application.device.OutboxDispatchResult; +import com.ruoyi.wcs.application.device.SnapshotCollector; +import com.ruoyi.wcs.application.dispatch.DispatchCoordinator; +import com.ruoyi.wcs.application.dispatch.DispatchCycleResult; +import com.ruoyi.wcs.application.dispatch.DispatchSnapshot; +import com.ruoyi.wcs.application.dispatch.GrantedDispatch; +import com.ruoyi.wcs.application.recovery.RecoveryGate; +import com.ruoyi.wcs.domain.shared.WcsClock; +import com.ruoyi.wcs.port.out.DispatchSnapshotProvider; + +import java.time.Instant; +import java.util.Objects; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +public final class WcsRuntimeCoordinator { + private final SnapshotCollector snapshotCollector; + private final RecoveryGate recoveryGate; + private final Supplier snapshotLoader; + private final Function dispatcher; + private final Consumer commandCreator; + private final BiFunction outboxSender; + private final WcsClock clock; + private final int outboxBatchLimit; + + public WcsRuntimeCoordinator(SnapshotCollector snapshotCollector, + RecoveryGate recoveryGate, + DispatchSnapshotProvider snapshotProvider, + DispatchCoordinator dispatchCoordinator, + DeviceExecutionService executionService, + CommandOutboxDispatcher outboxDispatcher, + WcsClock clock, int outboxBatchLimit) { + this(snapshotCollector, recoveryGate, + Objects.requireNonNull(snapshotProvider, + "snapshotProvider")::load, + Objects.requireNonNull(dispatchCoordinator, + "dispatchCoordinator")::dispatch, + grant -> Objects.requireNonNull(executionService, + "executionService").createCommand(grant), + (limit, now) -> Objects.requireNonNull(outboxDispatcher, + "outboxDispatcher").dispatchDue(limit, now), + clock, outboxBatchLimit); + } + + WcsRuntimeCoordinator(SnapshotCollector snapshotCollector, + RecoveryGate recoveryGate, + Supplier snapshotLoader, + Function dispatcher, + Consumer commandCreator, + BiFunction outboxSender, + WcsClock clock, int outboxBatchLimit) { + this.snapshotCollector = Objects.requireNonNull(snapshotCollector, + "snapshotCollector"); + this.recoveryGate = Objects.requireNonNull(recoveryGate, + "recoveryGate"); + this.snapshotLoader = Objects.requireNonNull(snapshotLoader, + "snapshotLoader"); + this.dispatcher = Objects.requireNonNull(dispatcher, "dispatcher"); + this.commandCreator = Objects.requireNonNull(commandCreator, + "commandCreator"); + this.outboxSender = Objects.requireNonNull(outboxSender, + "outboxSender"); + this.clock = Objects.requireNonNull(clock, "clock"); + if (outboxBatchLimit <= 0) { + throw new IllegalArgumentException( + "outboxBatchLimit must be positive"); + } + this.outboxBatchLimit = outboxBatchLimit; + } + + public RuntimeTickResult tick() { + snapshotCollector.collectAndAccept(); + if (!recoveryGate.allowsDispatch()) { + return RuntimeTickResult.recoveryBlocked(); + } + DispatchCycleResult dispatch = Objects.requireNonNull( + dispatcher.apply(Objects.requireNonNull(snapshotLoader.get(), + "snapshot provider returned null")), + "dispatch coordinator returned null"); + int createdCommands = 0; + for (GrantedDispatch grant : dispatch.getGranted()) { + commandCreator.accept(grant); + createdCommands++; + } + OutboxDispatchResult outbox = Objects.requireNonNull( + outboxSender.apply(outboxBatchLimit, clock.now()), + "outbox dispatcher returned null"); + return RuntimeTickResult.completed(dispatch, outbox, createdCommands); + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java index 5561bd5..d4f69c6 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/DeviceExecutionService.java @@ -11,7 +11,9 @@ import com.ruoyi.wcs.port.out.DeviceCommandRepository; import org.springframework.transaction.annotation.Transactional; import java.time.Instant; +import java.util.Map; import java.util.Objects; +import java.util.TreeMap; import java.util.function.Supplier; public final class DeviceExecutionService { @@ -69,10 +71,24 @@ public final class DeviceExecutionService { } route.append(requireSafe(node, "route node")); } - return "orderId=" + requireSafe(grant.getOrderId(), "orderId") + StringBuilder payload = new StringBuilder( + "orderId=" + requireSafe(grant.getOrderId(), "orderId") + "|stepId=" + requireSafe(grant.getStepId(), "stepId") + "|vehicleId=" + requireSafe(grant.getVehicleId(), "vehicleId") - + "|route=" + route; + + "|route=" + route); + for (Map.Entry entry : new TreeMap( + grant.getCommandParameters()).entrySet()) { + String key = requireSafe(entry.getKey(), "command parameter key"); + if ("orderId".equals(key) || "stepId".equals(key) + || "vehicleId".equals(key) || "route".equals(key)) { + throw new IllegalArgumentException( + "command parameter overrides reserved field " + key); + } + payload.append('|').append(key).append('=') + .append(requireSafe(entry.getValue(), + "command parameter " + key)); + } + return payload.toString(); } private static String requireSafe(String value, String field) { diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/FeedbackReconciler.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/FeedbackReconciler.java index 5f38024..7f32b88 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/FeedbackReconciler.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/FeedbackReconciler.java @@ -133,7 +133,9 @@ public final class FeedbackReconciler if (!execution.getTaskNumber().equals(snapshot.getTaskNumber())) { return recovery(snapshot, execution, "TASK_NUMBER_MISMATCH"); } - if (!execution.getExpectedBootId().equals(snapshot.getDeviceBootId())) { + if (snapshot.hasDeviceBootSession() + && !execution.getExpectedBootId().equals( + snapshot.getDeviceBootId())) { return recovery(snapshot, execution, "DEVICE_RESTARTED"); } if (execution.getTopologyVersion() != snapshot.getTopologyVersion()) { @@ -142,6 +144,11 @@ public final class FeedbackReconciler if (execution.getFencingToken() != snapshot.getFencingToken()) { return recovery(snapshot, execution, "FENCING_TOKEN_MISMATCH"); } + if (!snapshot.hasOrderedDeviceSequence()) { + return new ReconcileResult(ReconcileStatus.CONSISTENT, + execution.getCommandKey(), Collections.emptyList(), + "LEGACY_UNORDERED_DIAGNOSTIC_ONLY"); + } if (snapshot.getDeviceSequence() <= execution.getLastDeviceSequence()) { return new ReconcileResult(ReconcileStatus.IGNORED_STALE, execution.getCommandKey(), Collections.emptyList(), null); diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SnapshotCollector.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SnapshotCollector.java new file mode 100644 index 0000000..8561775 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/device/SnapshotCollector.java @@ -0,0 +1,6 @@ +package com.ruoyi.wcs.application.device; + +@FunctionalInterface +public interface SnapshotCollector { + void collectAndAccept(); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinator.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinator.java index 2666c38..b1dc6d1 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinator.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinator.java @@ -114,7 +114,9 @@ public final class DispatchCoordinator { granted.add(new GrantedDispatch(task.getOrderId(), task.getStepId(), assignment.getVehicleId(), assignment.getRoute(), Collections.singletonList(reservation.getReservationId()), - snapshot.getTopology().version(), reservation.getFencingToken())); + snapshot.getTopology().version(), + reservation.getFencingToken(), + task.getCommandParameters())); } else { HeldDispatch blocked = new HeldDispatch(task.getOrderId(), task.getStepId(), assignment.getVehicleId(), traffic.getType(), diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/GrantedDispatch.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/GrantedDispatch.java index e1c1a17..c6e3d83 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/GrantedDispatch.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/dispatch/GrantedDispatch.java @@ -4,7 +4,10 @@ import com.ruoyi.wcs.domain.traffic.RoutePlan; import java.util.ArrayList; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +import java.util.Objects; public final class GrantedDispatch { private final String orderId; @@ -14,10 +17,20 @@ public final class GrantedDispatch { private final List reservationIds; private final long topologyVersion; private final long fencingToken; + private final Map commandParameters; public GrantedDispatch(String orderId, String stepId, String vehicleId, RoutePlan protectedRoute, List reservationIds, long topologyVersion, long fencingToken) { + this(orderId, stepId, vehicleId, protectedRoute, reservationIds, + topologyVersion, fencingToken, + Collections.emptyMap()); + } + + public GrantedDispatch(String orderId, String stepId, String vehicleId, + RoutePlan protectedRoute, List reservationIds, + long topologyVersion, long fencingToken, + Map commandParameters) { this.orderId = orderId; this.stepId = stepId; this.vehicleId = vehicleId; @@ -26,6 +39,9 @@ public final class GrantedDispatch { new ArrayList(reservationIds)); this.topologyVersion = topologyVersion; this.fencingToken = fencingToken; + this.commandParameters = Collections.unmodifiableMap( + new LinkedHashMap(Objects.requireNonNull( + commandParameters, "commandParameters"))); } public String getOrderId() { @@ -55,4 +71,8 @@ public final class GrantedDispatch { public long getFencingToken() { return fencingToken; } + + public Map getCommandParameters() { + return commandParameters; + } } diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/recovery/RecoveryGate.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/recovery/RecoveryGate.java new file mode 100644 index 0000000..4f0fc5a --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/recovery/RecoveryGate.java @@ -0,0 +1,6 @@ +package com.ruoyi.wcs.application.recovery; + +@FunctionalInterface +public interface RecoveryGate { + boolean allowsDispatch(); +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/dispatch/DispatchTask.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/dispatch/DispatchTask.java index 2303478..1f8e005 100644 --- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/dispatch/DispatchTask.java +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/dispatch/DispatchTask.java @@ -4,6 +4,8 @@ import com.ruoyi.wcs.domain.equipment.DeviceCapability; import java.util.Collections; import java.util.EnumSet; +import java.util.LinkedHashMap; +import java.util.Map; import java.util.Objects; import java.util.Set; @@ -16,11 +18,22 @@ public final class DispatchTask { private final long waitingAgeSeconds; private final int minimumBattery; private final Set requiredCapabilities; + private final Map commandParameters; public DispatchTask(String orderId, String stepId, String sourceNodeId, String destinationNodeId, int priority, long waitingAgeSeconds, int minimumBattery, Set requiredCapabilities) { + this(orderId, stepId, sourceNodeId, destinationNodeId, priority, + waitingAgeSeconds, minimumBattery, requiredCapabilities, + Collections.emptyMap()); + } + + public DispatchTask(String orderId, String stepId, + String sourceNodeId, String destinationNodeId, + int priority, long waitingAgeSeconds, int minimumBattery, + Set requiredCapabilities, + Map commandParameters) { this.orderId = requireText(orderId, "orderId"); this.stepId = requireText(stepId, "stepId"); this.sourceNodeId = requireText(sourceNodeId, "sourceNodeId"); @@ -39,6 +52,7 @@ public final class DispatchTask { ? EnumSet.noneOf(DeviceCapability.class) : EnumSet.copyOf(requiredCapabilities); this.requiredCapabilities = Collections.unmodifiableSet(copied); + this.commandParameters = immutableParameters(commandParameters); } public String getOrderId() { @@ -73,6 +87,21 @@ public final class DispatchTask { return requiredCapabilities; } + public Map getCommandParameters() { + return commandParameters; + } + + private static Map immutableParameters( + Map source) { + Objects.requireNonNull(source, "commandParameters"); + Map copy = new LinkedHashMap(); + for (Map.Entry entry : source.entrySet()) { + copy.put(requireText(entry.getKey(), "command parameter key"), + requireText(entry.getValue(), "command parameter value")); + } + return Collections.unmodifiableMap(copy); + } + private static String requireText(String value, String field) { if (value == null || value.trim().isEmpty()) { throw new IllegalArgumentException(field + " must not be blank"); diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/WcsRuntimeCoordinatorTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/WcsRuntimeCoordinatorTest.java new file mode 100644 index 0000000..a89ff18 --- /dev/null +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/WcsRuntimeCoordinatorTest.java @@ -0,0 +1,125 @@ +package com.ruoyi.wcs.application; + +import com.ruoyi.wcs.application.device.OutboxDispatchResult; +import com.ruoyi.wcs.application.dispatch.DispatchCycleResult; +import com.ruoyi.wcs.application.dispatch.DispatchSnapshot; +import com.ruoyi.wcs.application.dispatch.GrantedDispatch; +import com.ruoyi.wcs.application.device.SnapshotCollector; +import com.ruoyi.wcs.application.recovery.RecoveryGate; +import com.ruoyi.wcs.domain.topology.EffectiveTopology; +import com.ruoyi.wcs.domain.topology.TopologyCompiler; +import com.ruoyi.wcs.domain.topology.TopologyDefinition; +import com.ruoyi.wcs.domain.topology.TopologyNode; +import com.ruoyi.wcs.domain.traffic.ResourceLedger; +import com.ruoyi.wcs.domain.traffic.RoutePlan; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +class WcsRuntimeCoordinatorTest { + + @Test + void tickRunsSnapshotRecoveryDispatchCommandAndOutboxInOrder() { + List calls = new ArrayList(); + SnapshotCollector snapshots = () -> calls.add("snapshot"); + RecoveryGate recovery = () -> { + calls.add("recovery"); + return true; + }; + GrantedDispatch grant = grant(); + DispatchCycleResult dispatchResult = DispatchCycleResult.committed( + Collections.singletonList(grant), + Collections.emptyList(), Collections.emptyList(), + Collections.emptyList()); + OutboxDispatchResult outboxResult = new OutboxDispatchResult(); + WcsRuntimeCoordinator coordinator = new WcsRuntimeCoordinator( + snapshots, recovery, + () -> { + calls.add("load"); + return snapshot(); + }, + value -> { + calls.add("dispatch"); + return dispatchResult; + }, + value -> calls.add("command:" + value.getStepId()), + (limit, now) -> { + calls.add("outbox:" + limit); + return outboxResult; + }, + () -> Instant.parse("2026-08-11T08:00:00Z"), 100); + + RuntimeTickResult result = coordinator.tick(); + + assertEquals(Arrays.asList("snapshot", "recovery", "load", + "dispatch", "command:STEP-1", "outbox:100"), calls); + assertEquals(RuntimeTickResult.Status.COMPLETED, result.getStatus()); + assertEquals(1, result.getCreatedCommandCount()); + assertEquals(dispatchResult, result.getDispatchResult()); + assertEquals(outboxResult, result.getOutboxResult()); + } + + @Test + void recoveryGateStopsBeforeDispatchAndOutbox() { + List calls = new ArrayList(); + WcsRuntimeCoordinator coordinator = new WcsRuntimeCoordinator( + () -> calls.add("snapshot"), + () -> { + calls.add("recovery"); + return false; + }, + () -> { + calls.add("load"); + return snapshot(); + }, + value -> { + calls.add("dispatch"); + return DispatchCycleResult.retrySnapshotChanged(); + }, + value -> calls.add("command"), + (limit, now) -> { + calls.add("outbox"); + return new OutboxDispatchResult(); + }, + Instant::now, 100); + + RuntimeTickResult result = coordinator.tick(); + + assertEquals(Arrays.asList("snapshot", "recovery"), calls); + assertEquals(RuntimeTickResult.Status.RECOVERY_BLOCKED, + result.getStatus()); + assertEquals(0, result.getCreatedCommandCount()); + assertNull(result.getDispatchResult()); + assertNull(result.getOutboxResult()); + } + + private static GrantedDispatch grant() { + RoutePlan route = new RoutePlan(Arrays.asList("A", "B"), + Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B"), + 1, 0, 7L); + return new GrantedDispatch("ORDER-1", "STEP-1", "VEH-1", route, + Collections.singletonList("RES-1"), 7L, 11L); + } + + private static DispatchSnapshot snapshot() { + EffectiveTopology topology = new TopologyCompiler().compile( + new TopologyDefinition("FT546", 7L, true, + Arrays.asList( + new TopologyNode("A", "F1", 0, 0, 1, + false, false), + new TopologyNode("B", "F1", 1, 0, 1, + false, false)), + Collections.emptyList(), Collections.emptyList(), + Collections.emptyList(), Collections.emptyList())); + return new DispatchSnapshot(Collections.emptyList(), + Collections.emptyList(), topology, + new ResourceLedger(topology), 1L); + } +} diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java index 75cd19a..97b18e2 100644 --- a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/CommandOutboxDispatcherTest.java @@ -68,6 +68,26 @@ class CommandOutboxDispatcherTest { assertEquals(0, gateway.sendCount); } + @Test + void explicitTaskNumberAndTypeReachTheDevicePayload() { + Map commandParameters = + new LinkedHashMap(); + commandParameters.put("taskNo", "30001"); + commandParameters.put("taskType", "1"); + RoutePlan route = new RoutePlan(Arrays.asList("A", "B"), + Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B"), + 1, 0, 7L); + GrantedDispatch grant = new GrantedDispatch("ORDER-1", "STEP-1", + "VEH-1", route, Collections.singletonList("RES-1"), 7L, + 11L, commandParameters); + + DeviceCommand command = service.createCommand(grant); + + assertEquals("orderId=ORDER-1|stepId=STEP-1|vehicleId=VEH-1" + + "|route=A>B|taskNo=30001|taskType=1", + command.getPayload()); + } + @Test void acceptedSendIsNotMarkedCompleted() { service.createCommand(grantedDispatch()); diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/FeedbackReconcilerTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/FeedbackReconcilerTest.java index c386f77..33663a2 100644 --- a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/FeedbackReconcilerTest.java +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/device/FeedbackReconcilerTest.java @@ -83,6 +83,24 @@ class FeedbackReconcilerTest { recovery.reasons); } + @Test + void unorderedSnapshotWithoutBootSessionIsDiagnosticOnly() { + DeviceSnapshot snapshot = DeviceSnapshot.unorderedWithoutBootSession( + "SNAP-LEGACY-1", "VEH-1", "CMD-KEY-1", "TASK-1", + at(5), true, 7L, 11L, "B", Arrays.asList("B", "C"), + false, Collections.emptyMap()); + + ReconcileResult result = reconciler.reconcile(snapshot); + + assertEquals(ReconcileStatus.CONSISTENT, result.getStatus()); + assertEquals("LEGACY_UNORDERED_DIAGNOSTIC_ONLY", + result.getRecoveryReason()); + assertTrue(recovery.reasons.isEmpty()); + assertEquals(0, reservations.releaseCallCount); + assertEquals(-1L, executions.findByCommandKey("CMD-KEY-1") + .getLastDeviceSequence()); + } + @Test void releasesOnlyResourcesStrictlyPassedByMatchingProgress() { DeviceSnapshot snapshot = snapshot("TASK-1", "B", diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinatorTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinatorTest.java index 5702444..85b8429 100644 --- a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinatorTest.java +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/dispatch/DispatchCoordinatorTest.java @@ -25,6 +25,8 @@ import java.time.Instant; import java.util.Arrays; import java.util.Collections; import java.util.EnumSet; +import java.util.LinkedHashMap; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -77,6 +79,27 @@ class DispatchCoordinatorTest { assertEquals(1, repository.commitCount); } + @Test + void carriesExplicitDeviceCommandParametersIntoTheGrant() { + EffectiveTopology topology = topology(); + Map commandParameters = + new LinkedHashMap(); + commandParameters.put("taskNo", "30001"); + commandParameters.put("taskType", "1"); + DispatchTask task = new DispatchTask("O1", "S1", "A", "X", 1, + 60L, 30, Collections.emptySet(), + commandParameters); + + DispatchCycleResult result = coordinator( + new RecordingTrafficRepository(true)).dispatch( + new DispatchSnapshot(Collections.singletonList(task), + Collections.singletonList(vehicle("V1", "A")), + topology, new ResourceLedger(topology), 11L)); + + assertEquals(commandParameters, + result.getGranted().get(0).getCommandParameters()); + } + private static DispatchCoordinator coordinator(TrafficStateRepository repository) { AStarRoutePlanner routePlanner = new AStarRoutePlanner(); AtomicInteger sequence = new AtomicInteger();