feat(ft546): run standard WCS as the single scheduler
This commit is contained in:
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<RuntimeTickResult> 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<RuntimeTickResult> 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();
|
||||
}
|
||||
}
|
||||
@@ -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<MonitoringThread>(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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
/**
|
||||
* 离开电桩
|
||||
*
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<DispatchSnapshot> snapshotLoader;
|
||||
private final Function<DispatchSnapshot, DispatchCycleResult> dispatcher;
|
||||
private final Consumer<GrantedDispatch> commandCreator;
|
||||
private final BiFunction<Integer, Instant, OutboxDispatchResult> 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<DispatchSnapshot> snapshotLoader,
|
||||
Function<DispatchSnapshot, DispatchCycleResult> dispatcher,
|
||||
Consumer<GrantedDispatch> commandCreator,
|
||||
BiFunction<Integer, Instant, OutboxDispatchResult> 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);
|
||||
}
|
||||
}
|
||||
@@ -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<String, String> entry : new TreeMap<String, String>(
|
||||
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) {
|
||||
|
||||
@@ -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.<String>emptyList(),
|
||||
"LEGACY_UNORDERED_DIAGNOSTIC_ONLY");
|
||||
}
|
||||
if (snapshot.getDeviceSequence() <= execution.getLastDeviceSequence()) {
|
||||
return new ReconcileResult(ReconcileStatus.IGNORED_STALE,
|
||||
execution.getCommandKey(), Collections.<String>emptyList(), null);
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
@FunctionalInterface
|
||||
public interface SnapshotCollector {
|
||||
void collectAndAccept();
|
||||
}
|
||||
@@ -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(),
|
||||
|
||||
@@ -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<String> reservationIds;
|
||||
private final long topologyVersion;
|
||||
private final long fencingToken;
|
||||
private final Map<String, String> commandParameters;
|
||||
|
||||
public GrantedDispatch(String orderId, String stepId, String vehicleId,
|
||||
RoutePlan protectedRoute, List<String> reservationIds,
|
||||
long topologyVersion, long fencingToken) {
|
||||
this(orderId, stepId, vehicleId, protectedRoute, reservationIds,
|
||||
topologyVersion, fencingToken,
|
||||
Collections.<String, String>emptyMap());
|
||||
}
|
||||
|
||||
public GrantedDispatch(String orderId, String stepId, String vehicleId,
|
||||
RoutePlan protectedRoute, List<String> reservationIds,
|
||||
long topologyVersion, long fencingToken,
|
||||
Map<String, String> commandParameters) {
|
||||
this.orderId = orderId;
|
||||
this.stepId = stepId;
|
||||
this.vehicleId = vehicleId;
|
||||
@@ -26,6 +39,9 @@ public final class GrantedDispatch {
|
||||
new ArrayList<String>(reservationIds));
|
||||
this.topologyVersion = topologyVersion;
|
||||
this.fencingToken = fencingToken;
|
||||
this.commandParameters = Collections.unmodifiableMap(
|
||||
new LinkedHashMap<String, String>(Objects.requireNonNull(
|
||||
commandParameters, "commandParameters")));
|
||||
}
|
||||
|
||||
public String getOrderId() {
|
||||
@@ -55,4 +71,8 @@ public final class GrantedDispatch {
|
||||
public long getFencingToken() {
|
||||
return fencingToken;
|
||||
}
|
||||
|
||||
public Map<String, String> getCommandParameters() {
|
||||
return commandParameters;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
package com.ruoyi.wcs.application.recovery;
|
||||
|
||||
@FunctionalInterface
|
||||
public interface RecoveryGate {
|
||||
boolean allowsDispatch();
|
||||
}
|
||||
@@ -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<DeviceCapability> requiredCapabilities;
|
||||
private final Map<String, String> commandParameters;
|
||||
|
||||
public DispatchTask(String orderId, String stepId,
|
||||
String sourceNodeId, String destinationNodeId,
|
||||
int priority, long waitingAgeSeconds, int minimumBattery,
|
||||
Set<DeviceCapability> requiredCapabilities) {
|
||||
this(orderId, stepId, sourceNodeId, destinationNodeId, priority,
|
||||
waitingAgeSeconds, minimumBattery, requiredCapabilities,
|
||||
Collections.<String, String>emptyMap());
|
||||
}
|
||||
|
||||
public DispatchTask(String orderId, String stepId,
|
||||
String sourceNodeId, String destinationNodeId,
|
||||
int priority, long waitingAgeSeconds, int minimumBattery,
|
||||
Set<DeviceCapability> requiredCapabilities,
|
||||
Map<String, String> 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<String, String> getCommandParameters() {
|
||||
return commandParameters;
|
||||
}
|
||||
|
||||
private static Map<String, String> immutableParameters(
|
||||
Map<String, String> source) {
|
||||
Objects.requireNonNull(source, "commandParameters");
|
||||
Map<String, String> copy = new LinkedHashMap<String, String>();
|
||||
for (Map.Entry<String, String> 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");
|
||||
|
||||
@@ -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<String> calls = new ArrayList<String>();
|
||||
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<String> calls = new ArrayList<String>();
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -68,6 +68,26 @@ class CommandOutboxDispatcherTest {
|
||||
assertEquals(0, gateway.sendCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
void explicitTaskNumberAndTypeReachTheDevicePayload() {
|
||||
Map<String, String> commandParameters =
|
||||
new LinkedHashMap<String, String>();
|
||||
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());
|
||||
|
||||
@@ -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.<String, String>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",
|
||||
|
||||
@@ -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<String, String> commandParameters =
|
||||
new LinkedHashMap<String, String>();
|
||||
commandParameters.put("taskNo", "30001");
|
||||
commandParameters.put("taskType", "1");
|
||||
DispatchTask task = new DispatchTask("O1", "S1", "A", "X", 1,
|
||||
60L, 30, Collections.<DeviceCapability>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();
|
||||
|
||||
Reference in New Issue
Block a user