test(ft546): verify standard WCS end to end

This commit is contained in:
zgw
2026-08-12 08:27:40 +08:00
parent ceba465675
commit f66585f717
31 changed files with 2560 additions and 16 deletions

View File

@@ -2,15 +2,31 @@ package com.ruoyi.project.ft546.runtime;
import com.ruoyi.wcs.application.WcsRuntimeCoordinator; import com.ruoyi.wcs.application.WcsRuntimeCoordinator;
import com.ruoyi.wcs.application.device.CommandOutboxDispatcher; import com.ruoyi.wcs.application.device.CommandOutboxDispatcher;
import com.ruoyi.wcs.application.device.CommandDeliveryRecorder;
import com.ruoyi.wcs.application.device.DeviceExecutionService; import com.ruoyi.wcs.application.device.DeviceExecutionService;
import com.ruoyi.wcs.application.device.FeedbackReconciler;
import com.ruoyi.wcs.application.device.SnapshotCollector; import com.ruoyi.wcs.application.device.SnapshotCollector;
import com.ruoyi.wcs.application.device.WmsNotificationOutboxDispatcher;
import com.ruoyi.wcs.application.device.WmsNotificationOutboxProjector;
import com.ruoyi.wcs.application.dispatch.DispatchCoordinator; import com.ruoyi.wcs.application.dispatch.DispatchCoordinator;
import com.ruoyi.wcs.application.recovery.RecoveryGate; import com.ruoyi.wcs.application.recovery.RecoveryGate;
import com.ruoyi.wcs.application.recovery.RecoveryApplicationService; import com.ruoyi.wcs.application.recovery.RecoveryApplicationService;
import com.ruoyi.wcs.application.task.TransportOrderExecutionService;
import com.ruoyi.wcs.domain.shared.IdGenerator; import com.ruoyi.wcs.domain.shared.IdGenerator;
import com.ruoyi.wcs.domain.shared.WcsClock; import com.ruoyi.wcs.domain.shared.WcsClock;
import com.ruoyi.wcs.port.out.CommandOutboxRepository;
import com.ruoyi.wcs.port.out.DeviceCommandGateway;
import com.ruoyi.wcs.port.out.DeviceCommandRepository;
import com.ruoyi.wcs.port.out.DeviceEventRepository;
import com.ruoyi.wcs.port.out.DeviceExecutionRepository;
import com.ruoyi.wcs.port.out.DispatchSnapshotProvider; import com.ruoyi.wcs.port.out.DispatchSnapshotProvider;
import com.ruoyi.wcs.port.out.DomainEventPublisher;
import com.ruoyi.wcs.port.out.RecoveryCaseRepository; import com.ruoyi.wcs.port.out.RecoveryCaseRepository;
import com.ruoyi.wcs.port.out.RecoveryCaseOpener;
import com.ruoyi.wcs.port.out.ResourceReservationRepository;
import com.ruoyi.wcs.port.out.TransportOrderExecutionRepository;
import com.ruoyi.wcs.port.out.WmsNotificationGateway;
import com.ruoyi.wcs.port.out.WmsNotificationOutboxRepository;
import com.ruoyi.wcs.query.WcsOperationsQueryService; import com.ruoyi.wcs.query.WcsOperationsQueryService;
import com.ruoyi.wcs.query.WcsOperationsReadRepository; import com.ruoyi.wcs.query.WcsOperationsReadRepository;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -23,6 +39,71 @@ import org.springframework.context.annotation.Configuration;
public class Ft546AdapterConfiguration { public class Ft546AdapterConfiguration {
private static final int OUTBOX_BATCH_LIMIT = 100; private static final int OUTBOX_BATCH_LIMIT = 100;
@Bean
public WmsNotificationOutboxProjector wmsNotificationOutboxProjector(
WmsNotificationOutboxRepository repository) {
return new WmsNotificationOutboxProjector(repository);
}
@Bean
public TransportOrderExecutionService transportOrderExecutionService(
TransportOrderExecutionRepository repository,
DomainEventPublisher eventPublisher,
WmsNotificationOutboxProjector notificationProjector) {
return new TransportOrderExecutionService(repository, eventPublisher,
notificationProjector);
}
@Bean
public DeviceExecutionService deviceExecutionService(
DeviceCommandRepository commandRepository,
CommandOutboxRepository outboxRepository,
DeviceExecutionRepository executionRepository,
TransportOrderExecutionService lifecycle,
IdGenerator idGenerator, WcsClock clock) {
return new DeviceExecutionService(commandRepository, outboxRepository,
executionRepository, lifecycle, idGenerator, clock::now);
}
@Bean
public FeedbackReconciler feedbackReconciler(
DeviceCommandRepository commandRepository,
DeviceExecutionRepository executionRepository,
DeviceEventRepository eventRepository,
ResourceReservationRepository reservationRepository,
RecoveryCaseOpener recoveryCaseOpener,
TransportOrderExecutionService lifecycle) {
return new FeedbackReconciler(commandRepository, executionRepository,
eventRepository, reservationRepository, recoveryCaseOpener,
lifecycle);
}
@Bean
public CommandDeliveryRecorder commandDeliveryRecorder(
DeviceCommandRepository commandRepository,
CommandOutboxRepository outboxRepository,
RecoveryCaseOpener recoveryCaseOpener,
ResourceReservationRepository reservationRepository) {
return new CommandDeliveryRecorder(commandRepository, outboxRepository,
recoveryCaseOpener, reservationRepository);
}
@Bean
public CommandOutboxDispatcher commandOutboxDispatcher(
CommandOutboxRepository outboxRepository,
DeviceCommandGateway commandGateway,
CommandDeliveryRecorder recorder) {
return new CommandOutboxDispatcher(outboxRepository, commandGateway,
recorder);
}
@Bean
public WmsNotificationOutboxDispatcher wmsNotificationOutboxDispatcher(
WmsNotificationOutboxRepository repository,
WmsNotificationGateway gateway) {
return new WmsNotificationOutboxDispatcher(repository, gateway);
}
@Bean @Bean
public WcsRuntimeCoordinator ft546WcsRuntimeCoordinator( public WcsRuntimeCoordinator ft546WcsRuntimeCoordinator(
SnapshotCollector snapshotCollector, SnapshotCollector snapshotCollector,
@@ -39,8 +120,11 @@ public class Ft546AdapterConfiguration {
@Bean @Bean
public Ft546WcsScheduler ft546WcsScheduler( public Ft546WcsScheduler ft546WcsScheduler(
WcsRuntimeCoordinator coordinator) { WcsRuntimeCoordinator coordinator,
return new Ft546WcsScheduler(coordinator); WmsNotificationOutboxDispatcher notificationDispatcher,
WcsClock clock) {
return new Ft546WcsScheduler(coordinator, notificationDispatcher,
clock, OUTBOX_BATCH_LIMIT);
} }
@Bean @Bean

View File

@@ -2,6 +2,8 @@ package com.ruoyi.project.ft546.runtime;
import com.ruoyi.wcs.application.RuntimeTickResult; import com.ruoyi.wcs.application.RuntimeTickResult;
import com.ruoyi.wcs.application.WcsRuntimeCoordinator; import com.ruoyi.wcs.application.WcsRuntimeCoordinator;
import com.ruoyi.wcs.application.device.WmsNotificationOutboxDispatcher;
import com.ruoyi.wcs.domain.shared.WcsClock;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
import java.util.Objects; import java.util.Objects;
@@ -11,16 +13,34 @@ import java.util.function.Supplier;
public final class Ft546WcsScheduler { public final class Ft546WcsScheduler {
private final Supplier<RuntimeTickResult> tick; private final Supplier<RuntimeTickResult> tick;
private final Runnable notificationDispatch;
private final AtomicBoolean tickInProgress = new AtomicBoolean(); private final AtomicBoolean tickInProgress = new AtomicBoolean();
private final AtomicLong completedTickCount = new AtomicLong(); private final AtomicLong completedTickCount = new AtomicLong();
private volatile RuntimeTickResult lastResult; private volatile RuntimeTickResult lastResult;
public Ft546WcsScheduler(WcsRuntimeCoordinator coordinator) { public Ft546WcsScheduler(WcsRuntimeCoordinator coordinator) {
this(Objects.requireNonNull(coordinator, "coordinator")::tick); this(Objects.requireNonNull(coordinator, "coordinator")::tick,
() -> { });
}
public Ft546WcsScheduler(
WcsRuntimeCoordinator coordinator,
WmsNotificationOutboxDispatcher notificationDispatcher,
WcsClock clock, int outboxBatchLimit) {
this(Objects.requireNonNull(coordinator, "coordinator")::tick,
notificationDispatch(notificationDispatcher, clock,
outboxBatchLimit));
} }
Ft546WcsScheduler(Supplier<RuntimeTickResult> tick) { Ft546WcsScheduler(Supplier<RuntimeTickResult> tick) {
this(tick, () -> { });
}
Ft546WcsScheduler(Supplier<RuntimeTickResult> tick,
Runnable notificationDispatch) {
this.tick = Objects.requireNonNull(tick, "tick"); this.tick = Objects.requireNonNull(tick, "tick");
this.notificationDispatch = Objects.requireNonNull(
notificationDispatch, "notificationDispatch");
} }
@Scheduled(fixedDelayString = @Scheduled(fixedDelayString =
@@ -29,13 +49,28 @@ public final class Ft546WcsScheduler {
if (!tickInProgress.compareAndSet(false, true)) { if (!tickInProgress.compareAndSet(false, true)) {
return; return;
} }
Throwable failure = null;
try { try {
lastResult = Objects.requireNonNull(tick.get(), try {
"runtime coordinator returned null"); lastResult = Objects.requireNonNull(tick.get(),
completedTickCount.incrementAndGet(); "runtime coordinator returned null");
completedTickCount.incrementAndGet();
} catch (RuntimeException | Error coreFailure) {
failure = coreFailure;
}
try {
notificationDispatch.run();
} catch (RuntimeException | Error notificationFailure) {
if (failure == null) {
failure = notificationFailure;
} else if (failure != notificationFailure) {
failure.addSuppressed(notificationFailure);
}
}
} finally { } finally {
tickInProgress.set(false); tickInProgress.set(false);
} }
rethrow(failure);
} }
public long getCompletedTickCount() { public long getCompletedTickCount() {
@@ -49,4 +84,33 @@ public final class Ft546WcsScheduler {
public boolean isTickInProgress() { public boolean isTickInProgress() {
return tickInProgress.get(); return tickInProgress.get();
} }
private static void rethrow(Throwable failure) {
if (failure == null) {
return;
}
if (failure instanceof RuntimeException) {
throw (RuntimeException) failure;
}
throw (Error) failure;
}
private static int requirePositive(int value) {
if (value <= 0) {
throw new IllegalArgumentException(
"outboxBatchLimit must be positive");
}
return value;
}
private static Runnable notificationDispatch(
WmsNotificationOutboxDispatcher dispatcher, WcsClock clock,
int batchLimit) {
WmsNotificationOutboxDispatcher requiredDispatcher =
Objects.requireNonNull(dispatcher, "notificationDispatcher");
WcsClock requiredClock = Objects.requireNonNull(clock, "clock");
int requiredLimit = requirePositive(batchLimit);
return () -> requiredDispatcher.dispatchDue(requiredLimit,
requiredClock.now());
}
} }

View File

@@ -11,5 +11,7 @@ devices:
taskType: "4002" taskType: "4002"
path: "4004" path: "4004"
trigger: "4021" trigger: "4021"
crossFloorFlag: "4020"
crossFloorTarget: "4022"
status: "4068" status: "4068"
execution: "4060" execution: "4060"

View File

@@ -107,6 +107,40 @@ class Ft546StandardRuntimeTest {
assertEquals(1L, scheduler.getCompletedTickCount()); assertEquals(1L, scheduler.getCompletedTickCount());
} }
@Test
void wmsOutboxStillDispatchesWhenCoreTickFails() {
AtomicInteger notificationDispatches = new AtomicInteger();
Ft546WcsScheduler scheduler = new Ft546WcsScheduler(() -> {
throw new IllegalStateException("core tick failed");
}, notificationDispatches::incrementAndGet);
IllegalStateException failure = assertThrows(
IllegalStateException.class, scheduler::runScheduledTick);
assertEquals("core tick failed", failure.getMessage());
assertEquals(1, notificationDispatches.get());
assertEquals(0L, scheduler.getCompletedTickCount());
assertFalse(scheduler.isTickInProgress());
}
@Test
void coreTickFailureRemainsPrimaryWhenNotificationDispatchAlsoFails() {
Ft546WcsScheduler scheduler = new Ft546WcsScheduler(() -> {
throw new IllegalStateException("core tick failed");
}, () -> {
throw new IllegalArgumentException("notification dispatch failed");
});
IllegalStateException failure = assertThrows(
IllegalStateException.class, scheduler::runScheduledTick);
assertEquals("core tick failed", failure.getMessage());
assertEquals(1, failure.getSuppressed().length);
assertEquals("notification dispatch failed",
failure.getSuppressed()[0].getMessage());
assertFalse(scheduler.isTickInProgress());
}
@Test @Test
void legacyModeDoesNotRequireStandardRuntimePorts() { void legacyModeDoesNotRequireStandardRuntimePorts() {
new ApplicationContextRunner() new ApplicationContextRunner()

View File

@@ -6,6 +6,7 @@ import com.ruoyi.wcs.domain.shared.WcsClock;
import com.ruoyi.wcs.infrastructure.persistence.task.MyBatisTransportOrderRepository; import com.ruoyi.wcs.infrastructure.persistence.task.MyBatisTransportOrderRepository;
import com.ruoyi.wcs.port.in.TransportOrderUseCase; import com.ruoyi.wcs.port.in.TransportOrderUseCase;
import com.ruoyi.wcs.port.out.TransportOrderRepository; import com.ruoyi.wcs.port.out.TransportOrderRepository;
import com.ruoyi.wcs.port.out.TransportOrderExecutionRepository;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -28,6 +29,8 @@ class WcsCoreWiringTest {
instanceof TransportOrderApplicationService); instanceof TransportOrderApplicationService);
assertTrue(context.getBean(TransportOrderRepository.class) assertTrue(context.getBean(TransportOrderRepository.class)
instanceof MyBatisTransportOrderRepository); instanceof MyBatisTransportOrderRepository);
assertTrue(context.getBean(TransportOrderExecutionRepository.class)
instanceof MyBatisTransportOrderRepository);
assertTrue(context.containsBeanDefinition("transportOrderMapper")); assertTrue(context.containsBeanDefinition("transportOrderMapper"));
WcsClock clock = context.getBean(WcsClock.class); WcsClock clock = context.getBean(WcsClock.class);

View File

@@ -0,0 +1,48 @@
{
"endpoint": "/wcs2Wms/api/receiveInMydoJob",
"request": {
"taskNo": "30001",
"cargoSpaceId": "070502",
"targetCargoSpaceId": null,
"exitId": null,
"trayCode": "TRAY-30001",
"entranceId": "070501",
"inOrOut": "10",
"jobId": "JOB-30001",
"roadwayId": "R1",
"classId": "7",
"priority": 5
},
"warehouseCode": "FT546",
"topologyVersion": 7,
"snapshotVersion": 11,
"vehicle": {
"deviceId": "1",
"floorCode": "F7",
"locationNodeId": "070501",
"batteryPercent": 80
},
"nodes": [
{"nodeId": "070501", "floorCode": "F7", "x": 0, "y": 0},
{"nodeId": "070502", "floorCode": "F7", "x": 1, "y": 0}
],
"edges": [
{"from": "070501", "to": "070502", "distanceCost": 1, "capacity": 1}
],
"commandParameters": {
"taskNo": "30001",
"taskType": "1"
},
"expectedCommandPayloadSuffix": "|vehicleId=1|route=070501>070502|taskNo=30001|taskType=1",
"expectedProtectedResources": [
"NODE:070501",
"EDGE:070501:070502",
"NODE:070502"
],
"expectedPlcWrites": [
{"address": "4004", "kind": "INT32", "values": [70501, 70502, 0, 0, 0, 0, 0, 0]},
{"address": "4000", "kind": "INT32", "values": [30001]},
{"address": "4002", "kind": "INT16", "values": [1]},
{"address": "4021", "kind": "INT16", "values": [1]}
]
}

View File

@@ -0,0 +1,49 @@
{
"endpoint": "/wcs2Wms/api/receiveInMydoJob",
"request": {
"taskNo": "30003",
"cargoSpaceId": "070503",
"targetCargoSpaceId": null,
"exitId": null,
"trayCode": "TRAY-30003",
"entranceId": "070501",
"inOrOut": "10",
"jobId": "JOB-30003",
"roadwayId": "R1",
"classId": "7",
"priority": 5
},
"warehouseCode": "FT546",
"topologyVersion": 9,
"snapshotVersion": 13,
"vehicle": {
"deviceId": "1",
"floorCode": "F7",
"locationNodeId": "070501",
"batteryPercent": 80
},
"nodes": [
{"nodeId": "070501", "floorCode": "F7", "x": 0, "y": 0},
{"nodeId": "070503", "floorCode": "F7", "x": 1, "y": 0}
],
"edges": [
{"from": "070501", "to": "070503", "distanceCost": 1, "capacity": 1}
],
"commandParameters": {
"taskNo": "30003",
"taskType": "1",
"stage": "1"
},
"expectedCommandPayloadSuffix": "|vehicleId=1|route=070501>070503|stage=1|taskNo=30003|taskType=1",
"expectedProtectedResources": [
"NODE:070501",
"EDGE:070501:070503",
"NODE:070503"
],
"expectedPlcWrites": [
{"address": "4404", "kind": "INT32", "values": [70501, 70503, 0, 0, 0, 0, 0, 0]},
{"address": "4422", "kind": "INT16", "values": [1]},
{"address": "4402", "kind": "INT16", "values": [1]},
{"address": "4400", "kind": "INT32", "values": [30003]}
]
}

View File

@@ -0,0 +1,50 @@
{
"endpoint": "/wcs2Wms/api/receiveInMydoJob",
"request": {
"taskNo": "30004",
"cargoSpaceId": "070504",
"targetCargoSpaceId": null,
"exitId": null,
"trayCode": "TRAY-30004",
"entranceId": "070503",
"inOrOut": "10",
"jobId": "JOB-30004",
"roadwayId": "R1",
"classId": "7",
"priority": 5
},
"warehouseCode": "FT546",
"topologyVersion": 10,
"snapshotVersion": 14,
"vehicle": {
"deviceId": "1",
"floorCode": "F7",
"locationNodeId": "070503",
"batteryPercent": 80
},
"nodes": [
{"nodeId": "070503", "floorCode": "F7", "x": 0, "y": 0},
{"nodeId": "070504", "floorCode": "F7", "x": 1, "y": 0}
],
"edges": [
{"from": "070503", "to": "070504", "distanceCost": 1, "capacity": 1}
],
"commandParameters": {
"taskNo": "30004",
"taskType": "9",
"stage": "2"
},
"expectedCommandPayloadSuffix": "|vehicleId=1|route=070503>070504|stage=2|taskNo=30004|taskType=9",
"expectedProtectedResources": [
"NODE:070503",
"EDGE:070503:070504",
"NODE:070504"
],
"expectedPlcWrites": [
{"address": "4404", "kind": "INT32", "values": [70503, 70504, 0, 0, 0, 0, 0, 0]},
{"address": "4422", "kind": "INT16", "values": [2]},
{"address": "4423", "kind": "INT16", "values": [99]},
{"address": "4402", "kind": "INT16", "values": [9]},
{"address": "4400", "kind": "INT32", "values": [30004]}
]
}

View File

@@ -0,0 +1,59 @@
{
"endpoint": "/wcs2Wms/api/receiveOutMydoJob",
"request": {
"taskNo": "30002",
"cargoSpaceId": "070501",
"targetCargoSpaceId": null,
"exitId": "080501",
"trayCode": "TRAY-30002",
"entranceId": null,
"inOrOut": "20",
"jobId": "JOB-30002",
"roadwayId": "R1",
"classId": "7",
"priority": 6
},
"warehouseCode": "FT546",
"topologyVersion": 8,
"snapshotVersion": 12,
"vehicle": {
"deviceId": "1",
"floorCode": "F7",
"locationNodeId": "070501",
"batteryPercent": 82
},
"nodes": [
{"nodeId": "070501", "floorCode": "F7", "x": 0, "y": 0},
{"nodeId": "070507", "floorCode": "F7", "x": 1, "y": 0},
{"nodeId": "080507", "floorCode": "F8", "x": 1, "y": 0},
{"nodeId": "080501", "floorCode": "F8", "x": 2, "y": 0}
],
"edges": [
{"from": "070501", "to": "070507", "distanceCost": 1, "capacity": 1},
{"from": "070507", "to": "080507", "distanceCost": 1, "capacity": 1},
{"from": "080507", "to": "080501", "distanceCost": 1, "capacity": 1}
],
"commandParameters": {
"taskNo": "30002",
"taskType": "8",
"crossFloorTarget": "8"
},
"expectedCommandPayloadSuffix": "|vehicleId=1|route=070501>070507>080507>080501|crossFloorTarget=8|taskNo=30002|taskType=8",
"expectedProtectedResources": [
"NODE:070501",
"EDGE:070501:070507",
"NODE:070507",
"EDGE:070507:080507",
"NODE:080507",
"EDGE:080501:080507",
"NODE:080501"
],
"expectedPlcWrites": [
{"address": "4004", "kind": "INT32", "values": [70501, 70507, 80507, 80501, 0, 0, 0, 0]},
{"address": "4000", "kind": "INT32", "values": [30002]},
{"address": "4002", "kind": "INT16", "values": [8]},
{"address": "4021", "kind": "INT16", "values": [1]},
{"address": "4020", "kind": "INT16", "values": [1]},
{"address": "4022", "kind": "INT16", "values": [8]}
]
}

View File

@@ -13,7 +13,7 @@ import java.time.Instant;
import java.util.Collections; import java.util.Collections;
import java.util.Objects; import java.util.Objects;
public final class CommandDeliveryRecorder { public class CommandDeliveryRecorder {
private final DeviceCommandRepository commandRepository; private final DeviceCommandRepository commandRepository;
private final CommandOutboxRepository outboxRepository; private final CommandOutboxRepository outboxRepository;
private final RecoveryCaseOpener recoveryCaseOpener; private final RecoveryCaseOpener recoveryCaseOpener;

View File

@@ -0,0 +1,10 @@
package com.ruoyi.wcs.application.device;
import com.ruoyi.wcs.domain.execution.DeviceCommand;
@FunctionalInterface
public interface DeviceCommandCompletionHandler {
DeviceCommandCompletionHandler NO_OP = command -> { };
void onSucceeded(DeviceCommand command);
}

View File

@@ -2,12 +2,14 @@ package com.ruoyi.wcs.application.device;
import com.ruoyi.wcs.application.dispatch.GrantedDispatch; import com.ruoyi.wcs.application.dispatch.GrantedDispatch;
import com.ruoyi.wcs.domain.execution.CommandOutboxEntry; import com.ruoyi.wcs.domain.execution.CommandOutboxEntry;
import com.ruoyi.wcs.domain.execution.CommittedDeviceExecution;
import com.ruoyi.wcs.domain.execution.DeviceCommand; import com.ruoyi.wcs.domain.execution.DeviceCommand;
import com.ruoyi.wcs.domain.execution.DeviceCommandType; import com.ruoyi.wcs.domain.execution.DeviceCommandType;
import com.ruoyi.wcs.domain.shared.DomainRuleViolation; import com.ruoyi.wcs.domain.shared.DomainRuleViolation;
import com.ruoyi.wcs.domain.shared.IdGenerator; import com.ruoyi.wcs.domain.shared.IdGenerator;
import com.ruoyi.wcs.port.out.CommandOutboxRepository; import com.ruoyi.wcs.port.out.CommandOutboxRepository;
import com.ruoyi.wcs.port.out.DeviceCommandRepository; import com.ruoyi.wcs.port.out.DeviceCommandRepository;
import com.ruoyi.wcs.port.out.DeviceExecutionRepository;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import java.time.Instant; import java.time.Instant;
@@ -16,9 +18,11 @@ import java.util.Objects;
import java.util.TreeMap; import java.util.TreeMap;
import java.util.function.Supplier; import java.util.function.Supplier;
public final class DeviceExecutionService { public class DeviceExecutionService {
private final DeviceCommandRepository commandRepository; private final DeviceCommandRepository commandRepository;
private final CommandOutboxRepository outboxRepository; private final CommandOutboxRepository outboxRepository;
private final DeviceExecutionRepository executionRepository;
private final DispatchGrantHandler dispatchGrantHandler;
private final IdGenerator idGenerator; private final IdGenerator idGenerator;
private final Supplier<Instant> clock; private final Supplier<Instant> clock;
@@ -26,10 +30,32 @@ public final class DeviceExecutionService {
CommandOutboxRepository outboxRepository, CommandOutboxRepository outboxRepository,
IdGenerator idGenerator, IdGenerator idGenerator,
Supplier<Instant> clock) { Supplier<Instant> clock) {
this(commandRepository, outboxRepository, null,
DispatchGrantHandler.NO_OP, idGenerator, clock);
}
public DeviceExecutionService(DeviceCommandRepository commandRepository,
CommandOutboxRepository outboxRepository,
DeviceExecutionRepository executionRepository,
IdGenerator idGenerator,
Supplier<Instant> clock) {
this(commandRepository, outboxRepository, executionRepository,
DispatchGrantHandler.NO_OP, idGenerator, clock);
}
public DeviceExecutionService(DeviceCommandRepository commandRepository,
CommandOutboxRepository outboxRepository,
DeviceExecutionRepository executionRepository,
DispatchGrantHandler dispatchGrantHandler,
IdGenerator idGenerator,
Supplier<Instant> clock) {
this.commandRepository = Objects.requireNonNull(commandRepository, this.commandRepository = Objects.requireNonNull(commandRepository,
"commandRepository"); "commandRepository");
this.outboxRepository = Objects.requireNonNull(outboxRepository, this.outboxRepository = Objects.requireNonNull(outboxRepository,
"outboxRepository"); "outboxRepository");
this.executionRepository = executionRepository;
this.dispatchGrantHandler = Objects.requireNonNull(dispatchGrantHandler,
"dispatchGrantHandler");
this.idGenerator = Objects.requireNonNull(idGenerator, "idGenerator"); this.idGenerator = Objects.requireNonNull(idGenerator, "idGenerator");
this.clock = Objects.requireNonNull(clock, "clock"); this.clock = Objects.requireNonNull(clock, "clock");
} }
@@ -50,6 +76,8 @@ public final class DeviceExecutionService {
return existing; return existing;
} }
dispatchGrantHandler.onGranted(grant);
Instant now = Objects.requireNonNull(clock.get(), "clock returned null"); Instant now = Objects.requireNonNull(clock.get(), "clock returned null");
DeviceCommand command = DeviceCommand.pending( DeviceCommand command = DeviceCommand.pending(
requireText(idGenerator.nextId(), "generated commandId"), requireText(idGenerator.nextId(), "generated commandId"),
@@ -59,10 +87,25 @@ public final class DeviceExecutionService {
command.getCommandId(), command.getCommandKey(), payload, now, command.getCommandId(), command.getCommandKey(), payload, now,
grant.getTopologyVersion(), grant.getFencingToken()); grant.getTopologyVersion(), grant.getFencingToken());
commandRepository.save(command); commandRepository.save(command);
if (executionRepository != null) {
executionRepository.save(execution(command, grant));
}
outboxRepository.saveIfAbsent(entry); outboxRepository.saveIfAbsent(entry);
return command; return command;
} }
private static CommittedDeviceExecution execution(
DeviceCommand command, GrantedDispatch grant) {
return new CommittedDeviceExecution(command.getCommandId(),
command.getCommandKey(), command.getDeviceId(),
requireText(grant.getCommandParameters().get("taskNo"),
"taskNo"),
grant.getProtectedRoute().getNodeIds(),
grant.getProtectedRoute().getResourceKeys(),
grant.getTopologyVersion(), grant.getFencingToken(),
grant.getCommandParameters().get("expectedBootId"));
}
private static String encode(GrantedDispatch grant) { private static String encode(GrantedDispatch grant) {
StringBuilder route = new StringBuilder(); StringBuilder route = new StringBuilder();
for (String node : grant.getProtectedRoute().getNodeIds()) { for (String node : grant.getProtectedRoute().getNodeIds()) {

View File

@@ -0,0 +1,10 @@
package com.ruoyi.wcs.application.device;
import com.ruoyi.wcs.application.dispatch.GrantedDispatch;
@FunctionalInterface
public interface DispatchGrantHandler {
DispatchGrantHandler NO_OP = grant -> { };
void onGranted(GrantedDispatch grant);
}

View File

@@ -20,19 +20,31 @@ import java.time.Instant;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
public final class FeedbackReconciler public class FeedbackReconciler
implements DeviceEventUseCase, DeviceSnapshotUseCase { implements DeviceEventUseCase, DeviceSnapshotUseCase {
private final DeviceCommandRepository commandRepository; private final DeviceCommandRepository commandRepository;
private final DeviceExecutionRepository executionRepository; private final DeviceExecutionRepository executionRepository;
private final DeviceEventRepository eventRepository; private final DeviceEventRepository eventRepository;
private final ResourceReservationRepository reservationRepository; private final ResourceReservationRepository reservationRepository;
private final RecoveryCaseOpener recoveryCaseOpener; private final RecoveryCaseOpener recoveryCaseOpener;
private final DeviceCommandCompletionHandler completionHandler;
public FeedbackReconciler(DeviceCommandRepository commandRepository, public FeedbackReconciler(DeviceCommandRepository commandRepository,
DeviceExecutionRepository executionRepository, DeviceExecutionRepository executionRepository,
DeviceEventRepository eventRepository, DeviceEventRepository eventRepository,
ResourceReservationRepository reservationRepository, ResourceReservationRepository reservationRepository,
RecoveryCaseOpener recoveryCaseOpener) { RecoveryCaseOpener recoveryCaseOpener) {
this(commandRepository, executionRepository, eventRepository,
reservationRepository, recoveryCaseOpener,
DeviceCommandCompletionHandler.NO_OP);
}
public FeedbackReconciler(DeviceCommandRepository commandRepository,
DeviceExecutionRepository executionRepository,
DeviceEventRepository eventRepository,
ResourceReservationRepository reservationRepository,
RecoveryCaseOpener recoveryCaseOpener,
DeviceCommandCompletionHandler completionHandler) {
this.commandRepository = java.util.Objects.requireNonNull( this.commandRepository = java.util.Objects.requireNonNull(
commandRepository, "commandRepository"); commandRepository, "commandRepository");
this.executionRepository = java.util.Objects.requireNonNull( this.executionRepository = java.util.Objects.requireNonNull(
@@ -43,6 +55,8 @@ public final class FeedbackReconciler
reservationRepository, "reservationRepository"); reservationRepository, "reservationRepository");
this.recoveryCaseOpener = java.util.Objects.requireNonNull( this.recoveryCaseOpener = java.util.Objects.requireNonNull(
recoveryCaseOpener, "recoveryCaseOpener"); recoveryCaseOpener, "recoveryCaseOpener");
this.completionHandler = java.util.Objects.requireNonNull(
completionHandler, "completionHandler");
} }
@Override @Override
@@ -70,6 +84,7 @@ public final class FeedbackReconciler
} }
DeviceCommand command = requireCommand(execution.getCommandKey()); DeviceCommand command = requireCommand(execution.getCommandKey());
boolean completed = false;
try { try {
switch (event.getEventType()) { switch (event.getEventType()) {
case ACKNOWLEDGED: case ACKNOWLEDGED:
@@ -91,6 +106,7 @@ public final class FeedbackReconciler
event.getCurrentNodeId(), event.getRemainingRouteNodes(), event.getCurrentNodeId(), event.getRemainingRouteNodes(),
true, event.getEventId()); true, event.getEventId());
complete(command, event.getEventId(), event.getReceivedTime()); complete(command, event.getEventId(), event.getReceivedTime());
completed = true;
break; break;
case REJECTED: case REJECTED:
requireActive(command, false); requireActive(command, false);
@@ -113,6 +129,9 @@ public final class FeedbackReconciler
return EventAcceptanceResult.RECOVERY_REQUIRED; return EventAcceptanceResult.RECOVERY_REQUIRED;
} }
commandRepository.save(command); commandRepository.save(command);
if (completed) {
completionHandler.onSucceeded(command);
}
return EventAcceptanceResult.APPLIED; return EventAcceptanceResult.APPLIED;
} }
@@ -134,8 +153,9 @@ public final class FeedbackReconciler
return recovery(snapshot, execution, "TASK_NUMBER_MISMATCH"); return recovery(snapshot, execution, "TASK_NUMBER_MISMATCH");
} }
if (snapshot.hasDeviceBootSession() if (snapshot.hasDeviceBootSession()
&& !execution.getExpectedBootId().equals( && (!execution.hasExpectedBootSession()
snapshot.getDeviceBootId())) { || !execution.getExpectedBootId().equals(
snapshot.getDeviceBootId()))) {
return recovery(snapshot, execution, "DEVICE_RESTARTED"); return recovery(snapshot, execution, "DEVICE_RESTARTED");
} }
if (execution.getTopologyVersion() != snapshot.getTopologyVersion()) { if (execution.getTopologyVersion() != snapshot.getTopologyVersion()) {
@@ -174,6 +194,7 @@ public final class FeedbackReconciler
if (snapshot.isCommandCompleted()) { if (snapshot.isCommandCompleted()) {
complete(command, snapshot.getSnapshotId(), snapshot.getObservedAt()); complete(command, snapshot.getSnapshotId(), snapshot.getObservedAt());
commandRepository.save(command); commandRepository.save(command);
completionHandler.onSucceeded(command);
} }
return new ReconcileResult(released.isEmpty() return new ReconcileResult(released.isEmpty()
? ReconcileStatus.CONSISTENT : ReconcileStatus.ADVANCED, ? ReconcileStatus.CONSISTENT : ReconcileStatus.ADVANCED,

View File

@@ -8,7 +8,7 @@ import java.time.Instant;
import java.util.Objects; import java.util.Objects;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
public final class WmsNotificationOutboxProjector { public class WmsNotificationOutboxProjector {
private final WmsNotificationOutboxRepository repository; private final WmsNotificationOutboxRepository repository;
public WmsNotificationOutboxProjector( public WmsNotificationOutboxProjector(

View File

@@ -14,7 +14,7 @@ import java.util.List;
import java.util.Objects; import java.util.Objects;
import java.util.function.Supplier; import java.util.function.Supplier;
public final class RecoveryApplicationService public class RecoveryApplicationService
implements RecoveryUseCase, RecoveryCaseOpener { implements RecoveryUseCase, RecoveryCaseOpener {
private final RecoveryCaseRepository repository; private final RecoveryCaseRepository repository;
private final IdGenerator idGenerator; private final IdGenerator idGenerator;

View File

@@ -16,7 +16,7 @@ import org.springframework.transaction.annotation.Transactional;
import java.util.List; import java.util.List;
import java.util.Objects; import java.util.Objects;
public final class TransportOrderApplicationService implements TransportOrderUseCase { public class TransportOrderApplicationService implements TransportOrderUseCase {
private final TransportOrderRepository repository; private final TransportOrderRepository repository;
private final WorkflowDefinitionProvider workflowDefinitions; private final WorkflowDefinitionProvider workflowDefinitions;
private final DomainEventPublisher eventPublisher; private final DomainEventPublisher eventPublisher;

View File

@@ -0,0 +1,166 @@
package com.ruoyi.wcs.application.task;
import com.ruoyi.wcs.application.device.DeviceCommandCompletionHandler;
import com.ruoyi.wcs.application.device.DispatchGrantHandler;
import com.ruoyi.wcs.application.device.WmsNotificationOutboxProjector;
import com.ruoyi.wcs.application.dispatch.GrantedDispatch;
import com.ruoyi.wcs.domain.execution.DeviceCommand;
import com.ruoyi.wcs.domain.shared.DomainEvent;
import com.ruoyi.wcs.domain.shared.DomainRuleViolation;
import com.ruoyi.wcs.domain.task.TaskStep;
import com.ruoyi.wcs.domain.task.TaskStepState;
import com.ruoyi.wcs.domain.task.TransportOrder;
import com.ruoyi.wcs.domain.task.TransportOrderExecution;
import com.ruoyi.wcs.domain.task.TransportOrderState;
import com.ruoyi.wcs.domain.task.WorkflowInstance;
import com.ruoyi.wcs.port.out.DomainEventPublisher;
import com.ruoyi.wcs.port.out.TransportOrderExecutionRepository;
import com.ruoyi.wcs.port.out.WmsNotification;
import org.springframework.transaction.annotation.Transactional;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
public class TransportOrderExecutionService
implements DispatchGrantHandler, DeviceCommandCompletionHandler {
private static final String COMMAND_PREFIX = "DISPATCH:";
private final TransportOrderExecutionRepository repository;
private final DomainEventPublisher eventPublisher;
private final WmsNotificationOutboxProjector notificationProjector;
public TransportOrderExecutionService(
TransportOrderExecutionRepository repository,
DomainEventPublisher eventPublisher,
WmsNotificationOutboxProjector notificationProjector) {
this.repository = Objects.requireNonNull(repository, "repository");
this.eventPublisher = Objects.requireNonNull(eventPublisher,
"eventPublisher");
this.notificationProjector = Objects.requireNonNull(
notificationProjector, "notificationProjector");
}
@Override
@Transactional
public void onGranted(GrantedDispatch grant) {
Objects.requireNonNull(grant, "grant");
TransportOrderExecution execution = requireExecution(
grant.getOrderId());
TransportOrder order = execution.getOrder();
WorkflowInstance workflow = execution.getWorkflow();
TaskStep step = workflow.stepById(grant.getStepId());
if (step.getState() == TaskStepState.RUNNING
&& grant.getVehicleId().equals(step.getVehicleId())) {
return;
}
if (step.getState() != TaskStepState.READY) {
throw new DomainRuleViolation(
"dispatch grant requires a READY task step");
}
if (order.getState() == TransportOrderState.PLANNED) {
order.start();
} else if (order.getState() != TransportOrderState.EXECUTING) {
throw new DomainRuleViolation(
"dispatch grant requires a PLANNED or EXECUTING order");
}
step.assign(grant.getVehicleId(), grant.getReservationIds());
step.start();
repository.save(execution);
publishAndClear(order, false);
}
@Override
@Transactional
public void onSucceeded(DeviceCommand command) {
Objects.requireNonNull(command, "command");
DispatchReference reference = reference(command.getCommandKey());
if (reference == null) {
return;
}
TransportOrderExecution execution = requireExecution(
reference.orderId);
TransportOrder order = execution.getOrder();
WorkflowInstance workflow = execution.getWorkflow();
TaskStep step = workflow.stepById(reference.stepId);
if (step.getState() == TaskStepState.SUCCEEDED
&& order.getState() == TransportOrderState.COMPLETED) {
return;
}
if (step.getState() != TaskStepState.RUNNING) {
throw new DomainRuleViolation(
"command completion requires a RUNNING task step");
}
step.succeed();
workflow.onStepSucceeded(step.getStepKey());
if (workflow.allRequiredStepsSucceeded()) {
order.complete();
}
repository.save(execution);
publishAndClear(order, true);
}
private TransportOrderExecution requireExecution(String orderId) {
TransportOrderExecution execution = repository.load(orderId);
if (execution == null) {
throw new DomainRuleViolation(
"transport order execution not found: " + orderId);
}
return execution;
}
private void publishAndClear(TransportOrder order,
boolean projectTerminalEvents) {
List<DomainEvent> events = order.getUncommittedEvents();
eventPublisher.publish(events);
if (projectTerminalEvents) {
for (DomainEvent event : events) {
if (isTerminal(event.getType())) {
notificationProjector.project(notification(order, event),
event.getOccurredAt());
}
}
}
order.clearUncommittedEvents();
}
private static WmsNotification notification(TransportOrder order,
DomainEvent event) {
return new WmsNotification(event.getEventId(),
order.getExternalRequestId(), order.getOrderId(),
event.getOccurredAt(), event.getType(),
order.getResultCode(), order.getStateReason(),
Collections.<String, String>emptyMap());
}
private static boolean isTerminal(String eventType) {
return "OrderCompleted".equals(eventType)
|| "OrderFailed".equals(eventType)
|| "OrderCancelled".equals(eventType);
}
private static DispatchReference reference(String commandKey) {
if (commandKey == null || !commandKey.startsWith(COMMAND_PREFIX)) {
return null;
}
String value = commandKey.substring(COMMAND_PREFIX.length());
int separator = value.indexOf(':');
if (separator <= 0 || separator == value.length() - 1) {
throw new DomainRuleViolation("invalid dispatch command key");
}
return new DispatchReference(value.substring(0, separator),
value.substring(separator + 1));
}
private static final class DispatchReference {
private final String orderId;
private final String stepId;
private DispatchReference(String orderId, String stepId) {
this.orderId = orderId;
this.stepId = stepId;
}
}
}

View File

@@ -55,7 +55,7 @@ public final class CommittedDeviceExecution {
} }
this.topologyVersion = topologyVersion; this.topologyVersion = topologyVersion;
this.fencingToken = fencingToken; this.fencingToken = fencingToken;
this.expectedBootId = requireText(expectedBootId, "expectedBootId"); this.expectedBootId = optionalText(expectedBootId);
this.currentNodeIndex = currentNodeIndex; this.currentNodeIndex = currentNodeIndex;
this.releasedResourceCount = releasedResourceCount; this.releasedResourceCount = releasedResourceCount;
this.lastDeviceSequence = lastDeviceSequence; this.lastDeviceSequence = lastDeviceSequence;
@@ -134,6 +134,7 @@ public final class CommittedDeviceExecution {
public long getTopologyVersion() { return topologyVersion; } public long getTopologyVersion() { return topologyVersion; }
public long getFencingToken() { return fencingToken; } public long getFencingToken() { return fencingToken; }
public String getExpectedBootId() { return expectedBootId; } public String getExpectedBootId() { return expectedBootId; }
public boolean hasExpectedBootSession() { return expectedBootId != null; }
public long getLastDeviceSequence() { return lastDeviceSequence; } public long getLastDeviceSequence() { return lastDeviceSequence; }
public static final class Progress { public static final class Progress {
@@ -209,4 +210,8 @@ public final class CommittedDeviceExecution {
} }
return value.trim(); return value.trim();
} }
private static String optionalText(String value) {
return value == null || value.trim().isEmpty() ? null : value.trim();
}
} }

View File

@@ -5,6 +5,7 @@ import com.ruoyi.wcs.domain.shared.DomainRuleViolation;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Objects;
public final class TaskStep { public final class TaskStep {
private final String stepId; private final String stepId;
@@ -28,6 +29,30 @@ public final class TaskStep {
} }
} }
public static TaskStep restore(String stepId, String stepKey,
TaskStepType type, boolean required,
TaskStepState state, String reasonCode,
String vehicleId,
List<String> reservationIds,
long stateVersion) {
if (stateVersion < 0L) {
throw new IllegalArgumentException(
"stateVersion must not be negative");
}
WorkflowStepDefinition definition = new WorkflowStepDefinition(
stepKey, type, required, true,
Collections.<String>emptyList());
TaskStep step = new TaskStep(stepId, definition);
step.state = Objects.requireNonNull(state, "state");
step.reasonCode = reasonCode;
step.vehicleId = vehicleId;
step.reservationIds = Collections.unmodifiableList(
new ArrayList<String>(Objects.requireNonNull(
reservationIds, "reservationIds")));
step.stateVersion = stateVersion;
return step;
}
public void markReady() { public void markReady() {
requireState(TaskStepState.WAITING); requireState(TaskStepState.WAITING);
transitionTo(TaskStepState.READY); transitionTo(TaskStepState.READY);

View File

@@ -0,0 +1,54 @@
package com.ruoyi.wcs.domain.task;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
public final class TransportOrderExecution {
private final TransportOrder order;
private final WorkflowInstance workflow;
private final long loadedOrderVersion;
private final Map<String, Long> loadedStepVersions;
private TransportOrderExecution(TransportOrder order,
WorkflowInstance workflow) {
this.order = Objects.requireNonNull(order, "order");
this.workflow = Objects.requireNonNull(workflow, "workflow");
if (!order.getOrderId().equals(workflow.getOrderId())) {
throw new IllegalArgumentException(
"workflow belongs to another transport order");
}
this.loadedOrderVersion = order.getStateVersion();
Map<String, Long> versions = new LinkedHashMap<String, Long>();
for (TaskStep step : workflow.getSteps()) {
versions.put(step.getStepId(), step.getStateVersion());
}
this.loadedStepVersions = Collections.unmodifiableMap(versions);
}
public static TransportOrderExecution loaded(TransportOrder order,
WorkflowInstance workflow) {
return new TransportOrderExecution(order, workflow);
}
public TransportOrder getOrder() {
return order;
}
public WorkflowInstance getWorkflow() {
return workflow;
}
public long getLoadedOrderVersion() {
return loadedOrderVersion;
}
public long getLoadedStepVersion(String stepId) {
Long version = loadedStepVersions.get(stepId);
if (version == null) {
throw new IllegalArgumentException("unknown loaded step " + stepId);
}
return version;
}
}

View File

@@ -32,6 +32,33 @@ public final class WorkflowInstance {
validateDependencies(); validateDependencies();
} }
private WorkflowInstance(String workflowId, String orderId,
List<TaskStep> restoredSteps,
Map<String, List<String>> dependencies) {
this.workflowId = requireText(workflowId, "workflowId");
this.orderId = requireText(orderId, "orderId");
this.steps = new LinkedHashMap<String, TaskStep>();
for (TaskStep step : restoredSteps) {
if (steps.put(step.getStepKey(), step) != null) {
throw new IllegalArgumentException(
"duplicate workflow step " + step.getStepKey());
}
}
if (steps.isEmpty()) {
throw new IllegalArgumentException("steps must not be empty");
}
this.mandatoryDependencies =
new LinkedHashMap<String, List<String>>();
for (String stepKey : steps.keySet()) {
List<String> values = dependencies.get(stepKey);
this.mandatoryDependencies.put(stepKey,
values == null ? Collections.<String>emptyList()
: Collections.unmodifiableList(
new ArrayList<String>(values)));
}
validateDependencies();
}
public static WorkflowInstance create(String workflowId, String orderId, public static WorkflowInstance create(String workflowId, String orderId,
WorkflowDefinition definition) { WorkflowDefinition definition) {
if (definition == null) { if (definition == null) {
@@ -40,6 +67,17 @@ public final class WorkflowInstance {
return new WorkflowInstance(workflowId, orderId, definition); return new WorkflowInstance(workflowId, orderId, definition);
} }
public static WorkflowInstance restore(String workflowId, String orderId,
List<TaskStep> steps,
Map<String, List<String>> dependencies) {
if (steps == null || dependencies == null) {
throw new IllegalArgumentException(
"restored workflow data must not be null");
}
return new WorkflowInstance(workflowId, orderId,
new ArrayList<TaskStep>(steps), dependencies);
}
public List<TaskStep> readySteps() { public List<TaskStep> readySteps() {
List<TaskStep> result = new ArrayList<TaskStep>(); List<TaskStep> result = new ArrayList<TaskStep>();
for (Map.Entry<String, TaskStep> entry : steps.entrySet()) { for (Map.Entry<String, TaskStep> entry : steps.entrySet()) {
@@ -73,6 +111,26 @@ public final class WorkflowInstance {
return step; return step;
} }
public TaskStep stepById(String stepId) {
String required = requireText(stepId, "stepId");
for (TaskStep step : steps.values()) {
if (required.equals(step.getStepId())) {
return step;
}
}
throw new IllegalArgumentException("unknown workflow step id " + stepId);
}
public boolean allRequiredStepsSucceeded() {
for (TaskStep step : steps.values()) {
if (step.isRequired()
&& step.getState() != TaskStepState.SUCCEEDED) {
return false;
}
}
return true;
}
public String getWorkflowId() { public String getWorkflowId() {
return workflowId; return workflowId;
} }

View File

@@ -2,15 +2,21 @@ package com.ruoyi.wcs.infrastructure.persistence.task;
import com.ruoyi.wcs.domain.task.TaskStep; import com.ruoyi.wcs.domain.task.TaskStep;
import com.ruoyi.wcs.domain.task.TransportOrder; import com.ruoyi.wcs.domain.task.TransportOrder;
import com.ruoyi.wcs.domain.task.TransportOrderExecution;
import com.ruoyi.wcs.domain.task.WorkflowInstance; import com.ruoyi.wcs.domain.task.WorkflowInstance;
import com.ruoyi.wcs.port.out.TransportOrderExecutionRepository;
import com.ruoyi.wcs.port.out.TransportOrderRepository; import com.ruoyi.wcs.port.out.TransportOrderRepository;
import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.dao.OptimisticLockingFailureException;
import java.util.List; import java.util.List;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import java.util.Objects; import java.util.Objects;
public final class MyBatisTransportOrderRepository implements TransportOrderRepository { public final class MyBatisTransportOrderRepository
implements TransportOrderRepository, TransportOrderExecutionRepository {
private final TransportOrderMapper mapper; private final TransportOrderMapper mapper;
public MyBatisTransportOrderRepository(TransportOrderMapper mapper) { public MyBatisTransportOrderRepository(TransportOrderMapper mapper) {
@@ -73,4 +79,66 @@ public final class MyBatisTransportOrderRepository implements TransportOrderRepo
"transport order state changed concurrently: " + order.getOrderId()); "transport order state changed concurrently: " + order.getOrderId());
} }
} }
@Override
public TransportOrderExecution load(String orderId) {
TransportOrderRow orderRow = mapper.selectByOrderId(orderId);
if (orderRow == null) {
return null;
}
WorkflowRow workflowRow = mapper.selectWorkflowByOrderId(orderId);
if (workflowRow == null) {
throw new IllegalStateException(
"transport order has no workflow: " + orderId);
}
if (!orderId.equals(workflowRow.getOrderId())) {
throw new IllegalStateException(
"workflow belongs to another transport order");
}
List<TaskStep> steps = new ArrayList<TaskStep>();
for (TaskStepRow row : mapper.selectTaskSteps(
workflowRow.getWorkflowId())) {
steps.add(row.toDomain());
}
Map<String, List<String>> dependencies =
new LinkedHashMap<String, List<String>>();
for (TaskDependencyRow row : mapper.selectTaskDependencies(
workflowRow.getWorkflowId())) {
List<String> values = dependencies.get(row.getStepKey());
if (values == null) {
values = new ArrayList<String>();
dependencies.put(row.getStepKey(), values);
}
values.add(row.getDependsOnStepKey());
}
WorkflowInstance workflow = WorkflowInstance.restore(
workflowRow.getWorkflowId(), workflowRow.getOrderId(),
steps, dependencies);
return TransportOrderExecution.loaded(orderRow.toDomain(), workflow);
}
@Override
public void save(TransportOrderExecution execution) {
Objects.requireNonNull(execution, "execution");
TransportOrder order = execution.getOrder();
if (order.getStateVersion() != execution.getLoadedOrderVersion()
&& mapper.updateState(TransportOrderRow.fromDomain(order),
execution.getLoadedOrderVersion()) != 1) {
throw new OptimisticLockingFailureException(
"transport order state changed concurrently: "
+ order.getOrderId());
}
for (TaskStep step : execution.getWorkflow().getSteps()) {
long expectedVersion = execution.getLoadedStepVersion(
step.getStepId());
if (step.getStateVersion() != expectedVersion
&& mapper.updateTaskStep(TaskStepRow.fromDomain(
execution.getWorkflow().getWorkflowId(), step),
expectedVersion) != 1) {
throw new OptimisticLockingFailureException(
"task step state changed concurrently: "
+ step.getStepId());
}
}
}
} }

View File

@@ -1,6 +1,10 @@
package com.ruoyi.wcs.infrastructure.persistence.task; package com.ruoyi.wcs.infrastructure.persistence.task;
import com.ruoyi.wcs.domain.task.TaskStep; import com.ruoyi.wcs.domain.task.TaskStep;
import com.ruoyi.wcs.domain.task.TaskStepState;
import com.ruoyi.wcs.domain.task.TaskStepType;
import java.util.Collections;
public final class TaskStepRow { public final class TaskStepRow {
private String stepId; private String stepId;
@@ -27,6 +31,13 @@ public final class TaskStepRow {
return row; return row;
} }
public TaskStep toDomain() {
return TaskStep.restore(stepId, stepKey,
TaskStepType.valueOf(stepType), required,
TaskStepState.valueOf(stepState), reasonCode, vehicleId,
Collections.<String>emptyList(), stateVersion);
}
public String getStepId() { public String getStepId() {
return stepId; return stepId;
} }

View File

@@ -3,6 +3,8 @@ package com.ruoyi.wcs.infrastructure.persistence.task;
import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Param;
import java.util.List;
@Mapper @Mapper
public interface TransportOrderMapper { public interface TransportOrderMapper {
TransportOrderRow selectByExternalRequestId( TransportOrderRow selectByExternalRequestId(
@@ -20,4 +22,15 @@ public interface TransportOrderMapper {
int insertTaskStep(TaskStepRow row); int insertTaskStep(TaskStepRow row);
int insertTaskDependency(TaskDependencyRow row); int insertTaskDependency(TaskDependencyRow row);
WorkflowRow selectWorkflowByOrderId(@Param("orderId") String orderId);
List<TaskStepRow> selectTaskSteps(
@Param("workflowId") String workflowId);
List<TaskDependencyRow> selectTaskDependencies(
@Param("workflowId") String workflowId);
int updateTaskStep(@Param("row") TaskStepRow row,
@Param("expectedVersion") long expectedVersion);
} }

View File

@@ -0,0 +1,9 @@
package com.ruoyi.wcs.port.out;
import com.ruoyi.wcs.domain.task.TransportOrderExecution;
public interface TransportOrderExecutionRepository {
TransportOrderExecution load(String orderId);
void save(TransportOrderExecution execution);
}

View File

@@ -21,6 +21,34 @@
<result property="updatedTime" column="updated_time"/> <result property="updatedTime" column="updated_time"/>
</resultMap> </resultMap>
<resultMap id="WorkflowRowMap"
type="com.ruoyi.wcs.infrastructure.persistence.task.WorkflowRow">
<id property="workflowId" column="workflow_id"/>
<result property="orderId" column="order_id"/>
<result property="workflowState" column="workflow_state"/>
<result property="stateVersion" column="state_version"/>
</resultMap>
<resultMap id="TaskStepRowMap"
type="com.ruoyi.wcs.infrastructure.persistence.task.TaskStepRow">
<id property="stepId" column="step_id"/>
<result property="workflowId" column="workflow_id"/>
<result property="stepKey" column="step_key"/>
<result property="stepType" column="step_type"/>
<result property="required" column="required_flag"/>
<result property="stepState" column="step_state"/>
<result property="reasonCode" column="reason_code"/>
<result property="vehicleId" column="vehicle_id"/>
<result property="stateVersion" column="state_version"/>
</resultMap>
<resultMap id="TaskDependencyRowMap"
type="com.ruoyi.wcs.infrastructure.persistence.task.TaskDependencyRow">
<id property="workflowId" column="workflow_id"/>
<result property="stepKey" column="step_key"/>
<result property="dependsOnStepKey" column="depends_on_step_key"/>
</resultMap>
<sql id="OrderColumns"> <sql id="OrderColumns">
order_id, external_request_id, order_type, load_id, source_code, order_id, external_request_id, order_type, load_id, source_code,
destination_code, priority, order_state, workflow_id, result_code, destination_code, priority, order_state, workflow_id, result_code,
@@ -88,4 +116,35 @@
#{workflowId}, #{stepKey}, #{dependsOnStepKey} #{workflowId}, #{stepKey}, #{dependsOnStepKey}
) )
</insert> </insert>
<select id="selectWorkflowByOrderId" resultMap="WorkflowRowMap">
SELECT workflow_id, order_id, workflow_state, state_version
FROM dbo.wcs_workflow_instance
WHERE order_id = #{orderId}
</select>
<select id="selectTaskSteps" resultMap="TaskStepRowMap">
SELECT step_id, workflow_id, step_key, step_type, required_flag,
step_state, reason_code, vehicle_id, state_version
FROM dbo.wcs_task_step
WHERE workflow_id = #{workflowId}
ORDER BY step_id
</select>
<select id="selectTaskDependencies" resultMap="TaskDependencyRowMap">
SELECT workflow_id, step_key, depends_on_step_key
FROM dbo.wcs_task_dependency
WHERE workflow_id = #{workflowId}
ORDER BY step_key, depends_on_step_key
</select>
<update id="updateTaskStep">
UPDATE dbo.wcs_task_step
SET step_state = #{row.stepState},
reason_code = #{row.reasonCode},
vehicle_id = #{row.vehicleId},
state_version = #{row.stateVersion}
WHERE step_id = #{row.stepId}
AND state_version = #{expectedVersion}
</update>
</mapper> </mapper>

View File

@@ -1,23 +1,77 @@
package com.ruoyi.wcs.infrastructure.persistence.task; package com.ruoyi.wcs.infrastructure.persistence.task;
import com.ruoyi.wcs.domain.task.TaskStepType; import com.ruoyi.wcs.domain.task.TaskStepType;
import com.ruoyi.wcs.domain.task.TaskStepState;
import com.ruoyi.wcs.domain.task.TransportOrder; import com.ruoyi.wcs.domain.task.TransportOrder;
import com.ruoyi.wcs.domain.task.TransportOrderExecution;
import com.ruoyi.wcs.domain.task.TransportOrderState;
import com.ruoyi.wcs.domain.task.TransportOrderType; import com.ruoyi.wcs.domain.task.TransportOrderType;
import com.ruoyi.wcs.domain.task.WorkflowDefinition; import com.ruoyi.wcs.domain.task.WorkflowDefinition;
import com.ruoyi.wcs.domain.task.WorkflowInstance; import com.ruoyi.wcs.domain.task.WorkflowInstance;
import com.ruoyi.wcs.domain.task.WorkflowStepDefinition; import com.ruoyi.wcs.domain.task.WorkflowStepDefinition;
import com.ruoyi.wcs.port.out.TransportOrderExecutionRepository;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.dao.OptimisticLockingFailureException;
import java.time.Instant; import java.time.Instant;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collections; import java.util.Collections;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
class MyBatisTransportOrderRepositoryTest { class MyBatisTransportOrderRepositoryTest {
@Test
void implementsTheOrderExecutionRepositoryContract() {
Object repository = new MyBatisTransportOrderRepository(
new RecordingMapper());
assertTrue(repository instanceof TransportOrderExecutionRepository);
}
@Test
void loadsAndOptimisticallySavesOrderAndWorkflowExecution() {
RecordingMapper mapper = new RecordingMapper();
TransportOrder order = order();
WorkflowInstance workflow = WorkflowInstance.create("WF-1", "ORDER-1",
new WorkflowDefinition(TransportOrderType.INBOUND,
Collections.singletonList(new WorkflowStepDefinition(
"MOVE", TaskStepType.MOVE, true, true,
Collections.<String>emptyList()))));
order.plan(workflow.getWorkflowId());
order.start();
workflow.readySteps().get(0).assign("VEH-1");
workflow.step("MOVE").start();
mapper.selected = TransportOrderRow.fromDomain(order);
mapper.selectedWorkflow = WorkflowRow.fromDomain(workflow);
mapper.selectedSteps = Collections.singletonList(
TaskStepRow.fromDomain(workflow.getWorkflowId(),
workflow.step("MOVE")));
Object candidate = new MyBatisTransportOrderRepository(mapper);
assertTrue(candidate instanceof TransportOrderExecutionRepository);
TransportOrderExecutionRepository repository =
(TransportOrderExecutionRepository) candidate;
TransportOrderExecution execution = repository.load("ORDER-1");
assertEquals(TransportOrderState.EXECUTING,
execution.getOrder().getState());
assertEquals(TaskStepState.RUNNING,
execution.getWorkflow().step("MOVE").getState());
execution.getWorkflow().step("MOVE").succeed();
execution.getOrder().complete();
repository.save(execution);
assertEquals(2L, mapper.expectedVersion);
assertEquals(3L, mapper.expectedStepVersion);
assertEquals(1, mapper.stepUpdates);
}
@Test @Test
void rejectsAStateUpdateWhenThePersistedVersionChanged() { void rejectsAStateUpdateWhenThePersistedVersionChanged() {
RecordingMapper mapper = new RecordingMapper(); RecordingMapper mapper = new RecordingMapper();
@@ -72,6 +126,12 @@ class MyBatisTransportOrderRepositoryTest {
private int workflowInserts; private int workflowInserts;
private int stepInserts; private int stepInserts;
private int dependencyInserts; private int dependencyInserts;
private WorkflowRow selectedWorkflow;
private List<TaskStepRow> selectedSteps = Collections.emptyList();
private List<TaskDependencyRow> selectedDependencies =
Collections.emptyList();
private long expectedStepVersion = -1L;
private int stepUpdates;
@Override @Override
public TransportOrderRow selectByExternalRequestId(String externalRequestId) { public TransportOrderRow selectByExternalRequestId(String externalRequestId) {
@@ -112,5 +172,28 @@ class MyBatisTransportOrderRepositoryTest {
dependencyInserts++; dependencyInserts++;
return 1; return 1;
} }
@Override
public WorkflowRow selectWorkflowByOrderId(String orderId) {
return selectedWorkflow;
}
@Override
public List<TaskStepRow> selectTaskSteps(String workflowId) {
return selectedSteps;
}
@Override
public List<TaskDependencyRow> selectTaskDependencies(
String workflowId) {
return selectedDependencies;
}
@Override
public int updateTaskStep(TaskStepRow row, long expectedVersion) {
expectedStepVersion = expectedVersion;
stepUpdates++;
return 1;
}
} }
} }

View File

@@ -29,5 +29,9 @@ class TransportOrderMapperXmlTest {
assertTrue(configuration.hasStatement(namespace + "insertWorkflow")); assertTrue(configuration.hasStatement(namespace + "insertWorkflow"));
assertTrue(configuration.hasStatement(namespace + "insertTaskStep")); assertTrue(configuration.hasStatement(namespace + "insertTaskStep"));
assertTrue(configuration.hasStatement(namespace + "insertTaskDependency")); assertTrue(configuration.hasStatement(namespace + "insertTaskDependency"));
assertTrue(configuration.hasStatement(namespace + "selectWorkflowByOrderId"));
assertTrue(configuration.hasStatement(namespace + "selectTaskSteps"));
assertTrue(configuration.hasStatement(namespace + "selectTaskDependencies"));
assertTrue(configuration.hasStatement(namespace + "updateTaskStep"));
} }
} }

View File

@@ -0,0 +1,120 @@
SET XACT_ABORT ON;
BEGIN TRANSACTION;
IF OBJECT_ID('dbo.sys_menu', 'U') IS NULL
BEGIN
THROW 50001, 'dbo.sys_menu is required before installing FT546 WCS menus', 1;
END;
DECLARE @wcs_menu_id bigint;
SELECT TOP (1) @wcs_menu_id = menu_id
FROM dbo.sys_menu
WHERE parent_id = 0
AND path = 'wcs'
AND menu_type = 'M'
ORDER BY menu_id;
IF @wcs_menu_id IS NULL
BEGIN
INSERT INTO dbo.sys_menu (
menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon,
create_by, create_time, update_by, update_time, remark)
VALUES (
N'WCS调度', 0, 5, 'wcs', NULL, NULL, 'Wcs',
1, 0, 'M', '0', '0', NULL, 'dashboard',
'admin', GETDATE(), '', NULL, N'FT546 WCS调度目录');
SET @wcs_menu_id = CONVERT(bigint, SCOPE_IDENTITY());
END;
IF NOT EXISTS (
SELECT 1 FROM dbo.sys_menu WHERE perms = 'wcs:order:list')
BEGIN
INSERT INTO dbo.sys_menu (
menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon,
create_by, create_time, update_by, update_time, remark)
VALUES (
N'任务监控', @wcs_menu_id, 1, 'order', 'wcs/order/index', NULL,
'WcsOrder', 1, 0, 'C', '0', '0', 'wcs:order:list', 'list',
'admin', GETDATE(), '', NULL, N'标准WCS任务监控');
END;
IF NOT EXISTS (
SELECT 1 FROM dbo.sys_menu WHERE perms = 'wcs:device:list')
BEGIN
INSERT INTO dbo.sys_menu (
menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon,
create_by, create_time, update_by, update_time, remark)
VALUES (
N'设备监控', @wcs_menu_id, 2, 'device', 'wcs/device/index', NULL,
'WcsDevice', 1, 0, 'C', '0', '0', 'wcs:device:list', 'monitor',
'admin', GETDATE(), '', NULL, N'标准WCS设备事实监控');
END;
IF NOT EXISTS (
SELECT 1 FROM dbo.sys_menu WHERE perms = 'wcs:traffic:list')
BEGIN
INSERT INTO dbo.sys_menu (
menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon,
create_by, create_time, update_by, update_time, remark)
VALUES (
N'交通资源', @wcs_menu_id, 3, 'traffic', 'wcs/traffic/index', NULL,
'WcsTraffic', 1, 0, 'C', '0', '0', 'wcs:traffic:list', 'tree',
'admin', GETDATE(), '', NULL, N'标准WCS资源占用与避让监控');
END;
IF NOT EXISTS (
SELECT 1 FROM dbo.sys_menu WHERE perms = 'wcs:recovery:list')
BEGIN
INSERT INTO dbo.sys_menu (
menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon,
create_by, create_time, update_by, update_time, remark)
VALUES (
N'恢复处置', @wcs_menu_id, 4, 'recovery', 'wcs/recovery/index', NULL,
'WcsRecovery', 1, 0, 'C', '0', '0', 'wcs:recovery:list', 'skill',
'admin', GETDATE(), '', NULL, N'标准WCS人工恢复处置');
END;
DECLARE @order_menu_id bigint;
SELECT TOP (1) @order_menu_id = menu_id
FROM dbo.sys_menu
WHERE perms = 'wcs:order:list'
ORDER BY menu_id;
IF NOT EXISTS (
SELECT 1 FROM dbo.sys_menu WHERE perms = 'wcs:order:control')
BEGIN
INSERT INTO dbo.sys_menu (
menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon,
create_by, create_time, update_by, update_time, remark)
VALUES (
N'任务控制', @order_menu_id, 1, '#', NULL, NULL, '',
1, 0, 'F', '0', '0', 'wcs:order:control', '#',
'admin', GETDATE(), '', NULL, N'暂停、恢复和取消标准WCS任务');
END;
DECLARE @recovery_menu_id bigint;
SELECT TOP (1) @recovery_menu_id = menu_id
FROM dbo.sys_menu
WHERE perms = 'wcs:recovery:list'
ORDER BY menu_id;
IF NOT EXISTS (
SELECT 1 FROM dbo.sys_menu WHERE perms = 'wcs:recovery:confirm')
BEGIN
INSERT INTO dbo.sys_menu (
menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon,
create_by, create_time, update_by, update_time, remark)
VALUES (
N'恢复确认', @recovery_menu_id, 1, '#', NULL, NULL, '',
1, 0, 'F', '0', '0', 'wcs:recovery:confirm', '#',
'admin', GETDATE(), '', NULL, N'确认或终止标准WCS恢复案例');
END;
COMMIT TRANSACTION;