feat(wcs): reconcile device facts with execution state
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
import com.ruoyi.wcs.domain.execution.DeviceCommand;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceCommandType;
|
||||
|
||||
public final class DeviceCommandEnvelope {
|
||||
private final String commandId;
|
||||
private final String commandKey;
|
||||
private final String deviceId;
|
||||
private final DeviceCommandType type;
|
||||
private final String payload;
|
||||
private final int attempt;
|
||||
private final long topologyVersion;
|
||||
private final long fencingToken;
|
||||
|
||||
public DeviceCommandEnvelope(String commandId, String commandKey,
|
||||
String deviceId, DeviceCommandType type,
|
||||
String payload, int attempt,
|
||||
long topologyVersion, long fencingToken) {
|
||||
this.commandId = requireText(commandId, "commandId");
|
||||
this.commandKey = requireText(commandKey, "commandKey");
|
||||
this.deviceId = requireText(deviceId, "deviceId");
|
||||
this.type = java.util.Objects.requireNonNull(type, "type");
|
||||
this.payload = requireText(payload, "payload");
|
||||
if (attempt <= 0 || topologyVersion <= 0L || fencingToken <= 0L) {
|
||||
throw new IllegalArgumentException("command envelope versions are invalid");
|
||||
}
|
||||
this.attempt = attempt;
|
||||
this.topologyVersion = topologyVersion;
|
||||
this.fencingToken = fencingToken;
|
||||
}
|
||||
|
||||
public static DeviceCommandEnvelope from(DeviceCommand command,
|
||||
long topologyVersion,
|
||||
long fencingToken) {
|
||||
return new DeviceCommandEnvelope(command.getCommandId(),
|
||||
command.getCommandKey(), command.getDeviceId(), command.getType(),
|
||||
command.getPayload(), command.getAttempt(), topologyVersion,
|
||||
fencingToken);
|
||||
}
|
||||
|
||||
public String getCommandId() { return commandId; }
|
||||
public String getCommandKey() { return commandKey; }
|
||||
public String getDeviceId() { return deviceId; }
|
||||
public DeviceCommandType getType() { return type; }
|
||||
public String getPayload() { return payload; }
|
||||
public int getAttempt() { return attempt; }
|
||||
public long getTopologyVersion() { return topologyVersion; }
|
||||
public long getFencingToken() { return fencingToken; }
|
||||
|
||||
private static String requireText(String value, String field) {
|
||||
if (value == null || value.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException(field + " must not be blank");
|
||||
}
|
||||
return value.trim();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
public enum EventAcceptanceResult {
|
||||
APPLIED,
|
||||
DUPLICATE,
|
||||
STALE,
|
||||
RECOVERY_REQUIRED
|
||||
}
|
||||
@@ -0,0 +1,259 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
import com.ruoyi.wcs.domain.execution.CommandResolutionEvidence;
|
||||
import com.ruoyi.wcs.domain.execution.CommittedDeviceExecution;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceCommand;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceCommandState;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceEvent;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceSnapshot;
|
||||
import com.ruoyi.wcs.domain.shared.DomainRuleViolation;
|
||||
import com.ruoyi.wcs.port.in.DeviceEventUseCase;
|
||||
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.RecoveryCaseOpener;
|
||||
import com.ruoyi.wcs.port.out.ResourceReservationRepository;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
public final class FeedbackReconciler implements DeviceEventUseCase {
|
||||
private final DeviceCommandRepository commandRepository;
|
||||
private final DeviceExecutionRepository executionRepository;
|
||||
private final DeviceEventRepository eventRepository;
|
||||
private final ResourceReservationRepository reservationRepository;
|
||||
private final RecoveryCaseOpener recoveryCaseOpener;
|
||||
|
||||
public FeedbackReconciler(DeviceCommandRepository commandRepository,
|
||||
DeviceExecutionRepository executionRepository,
|
||||
DeviceEventRepository eventRepository,
|
||||
ResourceReservationRepository reservationRepository,
|
||||
RecoveryCaseOpener recoveryCaseOpener) {
|
||||
this.commandRepository = java.util.Objects.requireNonNull(
|
||||
commandRepository, "commandRepository");
|
||||
this.executionRepository = java.util.Objects.requireNonNull(
|
||||
executionRepository, "executionRepository");
|
||||
this.eventRepository = java.util.Objects.requireNonNull(
|
||||
eventRepository, "eventRepository");
|
||||
this.reservationRepository = java.util.Objects.requireNonNull(
|
||||
reservationRepository, "reservationRepository");
|
||||
this.recoveryCaseOpener = java.util.Objects.requireNonNull(
|
||||
recoveryCaseOpener, "recoveryCaseOpener");
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public EventAcceptanceResult accept(DeviceEvent event) {
|
||||
if (!eventRepository.claim(event)) {
|
||||
return EventAcceptanceResult.DUPLICATE;
|
||||
}
|
||||
CommittedDeviceExecution execution = executionRepository
|
||||
.findByCommandKey(event.getCommandKey());
|
||||
if (execution == null || !execution.getDeviceId().equals(event.getDeviceId())) {
|
||||
openRecovery(event.getDeviceId(), event.getCommandKey(),
|
||||
"COMMAND_CONTEXT_MISMATCH", event.getEventId(),
|
||||
event.getReceivedTime());
|
||||
return EventAcceptanceResult.RECOVERY_REQUIRED;
|
||||
}
|
||||
if (event.getDeviceSequence() <= execution.getLastDeviceSequence()) {
|
||||
return EventAcceptanceResult.STALE;
|
||||
}
|
||||
if (!execution.getTaskNumber().equals(event.getTaskNumber())) {
|
||||
openRecovery(execution.getDeviceId(), execution.getCommandKey(),
|
||||
"TASK_NUMBER_MISMATCH", event.getEventId(),
|
||||
event.getReceivedTime());
|
||||
return EventAcceptanceResult.RECOVERY_REQUIRED;
|
||||
}
|
||||
|
||||
DeviceCommand command = requireCommand(execution.getCommandKey());
|
||||
try {
|
||||
switch (event.getEventType()) {
|
||||
case ACKNOWLEDGED:
|
||||
requireActive(command, false);
|
||||
command.acknowledge(event.getEventId(),
|
||||
transitionTime(command, event.getReceivedTime()));
|
||||
executionRepository.save(execution.observeSequence(
|
||||
event.getDeviceSequence()));
|
||||
break;
|
||||
case PROGRESS:
|
||||
requireActive(command, false);
|
||||
applyProgress(execution, event.getDeviceSequence(),
|
||||
event.getCurrentNodeId(), event.getRemainingRouteNodes(),
|
||||
false, event.getEventId());
|
||||
break;
|
||||
case COMPLETED:
|
||||
requireActive(command, true);
|
||||
applyProgress(execution, event.getDeviceSequence(),
|
||||
event.getCurrentNodeId(), event.getRemainingRouteNodes(),
|
||||
true, event.getEventId());
|
||||
complete(command, event.getEventId(), event.getReceivedTime());
|
||||
break;
|
||||
case REJECTED:
|
||||
requireActive(command, false);
|
||||
command.reject("DEVICE_REJECTED:" + event.getEventId(),
|
||||
transitionTime(command, event.getReceivedTime()));
|
||||
executionRepository.save(execution.observeSequence(
|
||||
event.getDeviceSequence()));
|
||||
break;
|
||||
case FAULT:
|
||||
openRecovery(execution.getDeviceId(), execution.getCommandKey(),
|
||||
"DEVICE_FAULT", event.getEventId(),
|
||||
event.getReceivedTime());
|
||||
return EventAcceptanceResult.RECOVERY_REQUIRED;
|
||||
default:
|
||||
throw new IllegalStateException("unsupported device event");
|
||||
}
|
||||
} catch (DomainRuleViolation ex) {
|
||||
openRecovery(execution.getDeviceId(), execution.getCommandKey(),
|
||||
reason(ex), event.getEventId(), event.getReceivedTime());
|
||||
return EventAcceptanceResult.RECOVERY_REQUIRED;
|
||||
}
|
||||
commandRepository.save(command);
|
||||
return EventAcceptanceResult.APPLIED;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public ReconcileResult reconcile(DeviceSnapshot snapshot) {
|
||||
CommittedDeviceExecution execution = executionRepository
|
||||
.findActiveByDeviceId(snapshot.getDeviceId());
|
||||
if (execution == null) {
|
||||
return recovery(snapshot, null, "ACTIVE_EXECUTION_NOT_FOUND");
|
||||
}
|
||||
if (!snapshot.isConnected()) {
|
||||
return recovery(snapshot, execution, "COMMUNICATION_LOST");
|
||||
}
|
||||
if (!execution.getCommandKey().equals(snapshot.getCommandKey())) {
|
||||
return recovery(snapshot, execution, "COMMAND_KEY_MISMATCH");
|
||||
}
|
||||
if (!execution.getTaskNumber().equals(snapshot.getTaskNumber())) {
|
||||
return recovery(snapshot, execution, "TASK_NUMBER_MISMATCH");
|
||||
}
|
||||
if (!execution.getExpectedBootId().equals(snapshot.getDeviceBootId())) {
|
||||
return recovery(snapshot, execution, "DEVICE_RESTARTED");
|
||||
}
|
||||
if (execution.getTopologyVersion() != snapshot.getTopologyVersion()) {
|
||||
return recovery(snapshot, execution, "TOPOLOGY_VERSION_MISMATCH");
|
||||
}
|
||||
if (execution.getFencingToken() != snapshot.getFencingToken()) {
|
||||
return recovery(snapshot, execution, "FENCING_TOKEN_MISMATCH");
|
||||
}
|
||||
if (snapshot.getDeviceSequence() <= execution.getLastDeviceSequence()) {
|
||||
return new ReconcileResult(ReconcileStatus.IGNORED_STALE,
|
||||
execution.getCommandKey(), Collections.<String>emptyList(), null);
|
||||
}
|
||||
|
||||
DeviceCommand command = requireCommand(execution.getCommandKey());
|
||||
try {
|
||||
if (command.getState() == DeviceCommandState.UNKNOWN
|
||||
&& !snapshot.isCommandCompleted()) {
|
||||
execution.evaluateProgress(snapshot.getDeviceSequence(),
|
||||
snapshot.getCurrentNodeId(),
|
||||
snapshot.getRemainingRouteNodes(), false);
|
||||
return new ReconcileResult(ReconcileStatus.CONSISTENT,
|
||||
execution.getCommandKey(),
|
||||
Collections.<String>emptyList(),
|
||||
"UNKNOWN_DIAGNOSTIC_ONLY");
|
||||
}
|
||||
requireActive(command, snapshot.isCommandCompleted());
|
||||
List<String> released = applyProgress(execution,
|
||||
snapshot.getDeviceSequence(), snapshot.getCurrentNodeId(),
|
||||
snapshot.getRemainingRouteNodes(), snapshot.isCommandCompleted(),
|
||||
snapshot.getSnapshotId());
|
||||
if (snapshot.isCommandCompleted()) {
|
||||
complete(command, snapshot.getSnapshotId(), snapshot.getObservedAt());
|
||||
commandRepository.save(command);
|
||||
}
|
||||
return new ReconcileResult(released.isEmpty()
|
||||
? ReconcileStatus.CONSISTENT : ReconcileStatus.ADVANCED,
|
||||
execution.getCommandKey(), released, null);
|
||||
} catch (DomainRuleViolation ex) {
|
||||
return recovery(snapshot, execution, reason(ex));
|
||||
}
|
||||
}
|
||||
|
||||
private List<String> applyProgress(CommittedDeviceExecution execution,
|
||||
long sequence, String currentNode,
|
||||
List<String> remaining, boolean completed,
|
||||
String evidenceId) {
|
||||
CommittedDeviceExecution.Progress progress = execution.evaluateProgress(
|
||||
sequence, currentNode, remaining, completed);
|
||||
if (progress.isStale()) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<String> released = progress.getReleasedResourceKeys();
|
||||
if (!released.isEmpty()) {
|
||||
reservationRepository.releasePassedResources(
|
||||
execution.getCommandKey(), released, evidenceId,
|
||||
execution.getFencingToken());
|
||||
}
|
||||
executionRepository.save(execution.advance(progress));
|
||||
return released;
|
||||
}
|
||||
|
||||
private static void requireActive(DeviceCommand command,
|
||||
boolean physicalCompletion) {
|
||||
DeviceCommandState state = command.getState();
|
||||
boolean allowed = state == DeviceCommandState.SENT
|
||||
|| state == DeviceCommandState.ACKNOWLEDGED
|
||||
|| (physicalCompletion && state == DeviceCommandState.UNKNOWN);
|
||||
if (!allowed) {
|
||||
throw new DomainRuleViolation("COMMAND_STATE_MISMATCH");
|
||||
}
|
||||
}
|
||||
|
||||
private void complete(DeviceCommand command, String evidenceId, Instant at) {
|
||||
Instant transitionAt = transitionTime(command, at);
|
||||
if (command.getState() == DeviceCommandState.UNKNOWN) {
|
||||
command.succeedAfterRecovery(
|
||||
CommandResolutionEvidence.physicalCompletion(
|
||||
command.getCommandKey(), evidenceId), transitionAt);
|
||||
} else {
|
||||
command.succeed(evidenceId, transitionAt);
|
||||
}
|
||||
}
|
||||
|
||||
private ReconcileResult recovery(DeviceSnapshot snapshot,
|
||||
CommittedDeviceExecution execution,
|
||||
String reason) {
|
||||
String commandKey = execution == null
|
||||
? snapshot.getCommandKey() : execution.getCommandKey();
|
||||
openRecovery(snapshot.getDeviceId(), commandKey, reason,
|
||||
snapshot.getSnapshotId(), snapshot.getObservedAt());
|
||||
return new ReconcileResult(ReconcileStatus.RECOVERY_REQUIRED,
|
||||
commandKey, Collections.<String>emptyList(), reason);
|
||||
}
|
||||
|
||||
private void openRecovery(String deviceId, String commandKey, String reason,
|
||||
String evidenceId, Instant evidenceTime) {
|
||||
DeviceCommand command = commandRepository.findByCommandKey(commandKey);
|
||||
if (command != null && (command.getState() == DeviceCommandState.SENT
|
||||
|| command.getState() == DeviceCommandState.ACKNOWLEDGED)) {
|
||||
command.markUnknown(reason, transitionTime(command, evidenceTime));
|
||||
commandRepository.save(command);
|
||||
}
|
||||
reservationRepository.protectForUnknown(commandKey, evidenceId);
|
||||
recoveryCaseOpener.open(deviceId, commandKey, reason,
|
||||
Collections.singletonList(evidenceId));
|
||||
}
|
||||
|
||||
private DeviceCommand requireCommand(String commandKey) {
|
||||
DeviceCommand command = commandRepository.findByCommandKey(commandKey);
|
||||
if (command == null) {
|
||||
throw new DomainRuleViolation("COMMAND_NOT_FOUND");
|
||||
}
|
||||
return command;
|
||||
}
|
||||
|
||||
private static Instant transitionTime(DeviceCommand command, Instant candidate) {
|
||||
return candidate.isBefore(command.getUpdatedAt())
|
||||
? command.getUpdatedAt() : candidate;
|
||||
}
|
||||
|
||||
private static String reason(DomainRuleViolation violation) {
|
||||
String message = violation.getMessage();
|
||||
return message == null || message.trim().isEmpty()
|
||||
? "RECONCILIATION_FAILED" : message;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
public final class ReconcileResult {
|
||||
private final ReconcileStatus status;
|
||||
private final String commandKey;
|
||||
private final List<String> releasedResourceKeys;
|
||||
private final String recoveryReason;
|
||||
|
||||
public ReconcileResult(ReconcileStatus status, String commandKey,
|
||||
List<String> releasedResourceKeys,
|
||||
String recoveryReason) {
|
||||
this.status = java.util.Objects.requireNonNull(status, "status");
|
||||
this.commandKey = commandKey;
|
||||
this.releasedResourceKeys = Collections.unmodifiableList(
|
||||
new ArrayList<String>(releasedResourceKeys));
|
||||
this.recoveryReason = recoveryReason;
|
||||
}
|
||||
|
||||
public ReconcileStatus getStatus() { return status; }
|
||||
public String getCommandKey() { return commandKey; }
|
||||
public List<String> getReleasedResourceKeys() {
|
||||
return releasedResourceKeys;
|
||||
}
|
||||
public String getRecoveryReason() { return recoveryReason; }
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
public enum ReconcileStatus {
|
||||
CONSISTENT,
|
||||
ADVANCED,
|
||||
IGNORED_STALE,
|
||||
RECOVERY_REQUIRED
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
public final class SendReceipt {
|
||||
public enum Status {
|
||||
ACCEPTED,
|
||||
ACCEPTED_UNKNOWN,
|
||||
NOT_SENT,
|
||||
REJECTED
|
||||
}
|
||||
|
||||
private final Status status;
|
||||
private final String receiptId;
|
||||
private final String detail;
|
||||
|
||||
public SendReceipt(Status status, String receiptId, String detail) {
|
||||
this.status = java.util.Objects.requireNonNull(status, "status");
|
||||
this.receiptId = receiptId;
|
||||
this.detail = detail;
|
||||
if ((status == Status.ACCEPTED || status == Status.ACCEPTED_UNKNOWN)
|
||||
&& isBlank(receiptId)) {
|
||||
throw new IllegalArgumentException(
|
||||
"accepted send receipt must have a receiptId");
|
||||
}
|
||||
}
|
||||
|
||||
public static SendReceipt accepted(String receiptId) {
|
||||
return new SendReceipt(Status.ACCEPTED, receiptId, null);
|
||||
}
|
||||
|
||||
public static SendReceipt acceptedUnknown(String receiptId, String detail) {
|
||||
return new SendReceipt(Status.ACCEPTED_UNKNOWN, receiptId, detail);
|
||||
}
|
||||
|
||||
public static SendReceipt notSent(String detail) {
|
||||
return new SendReceipt(Status.NOT_SENT, null, detail);
|
||||
}
|
||||
|
||||
public static SendReceipt rejected(String detail) {
|
||||
return new SendReceipt(Status.REJECTED, null, detail);
|
||||
}
|
||||
|
||||
public Status getStatus() { return status; }
|
||||
public String getReceiptId() { return receiptId; }
|
||||
public String getDetail() { return detail; }
|
||||
|
||||
private static boolean isBlank(String value) {
|
||||
return value == null || value.trim().isEmpty();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,212 @@
|
||||
package com.ruoyi.wcs.domain.execution;
|
||||
|
||||
import com.ruoyi.wcs.domain.shared.DomainRuleViolation;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
public final class CommittedDeviceExecution {
|
||||
private final String commandId;
|
||||
private final String commandKey;
|
||||
private final String deviceId;
|
||||
private final String taskNumber;
|
||||
private final List<String> routeNodeIds;
|
||||
private final List<String> routeResourceKeys;
|
||||
private final long topologyVersion;
|
||||
private final long fencingToken;
|
||||
private final String expectedBootId;
|
||||
private final int currentNodeIndex;
|
||||
private final int releasedResourceCount;
|
||||
private final long lastDeviceSequence;
|
||||
|
||||
public CommittedDeviceExecution(String commandId, String commandKey,
|
||||
String deviceId, String taskNumber,
|
||||
List<String> routeNodeIds,
|
||||
List<String> routeResourceKeys,
|
||||
long topologyVersion, long fencingToken,
|
||||
String expectedBootId) {
|
||||
this(commandId, commandKey, deviceId, taskNumber, routeNodeIds,
|
||||
routeResourceKeys, topologyVersion, fencingToken, expectedBootId,
|
||||
0, 0, -1L);
|
||||
}
|
||||
|
||||
private CommittedDeviceExecution(String commandId, String commandKey,
|
||||
String deviceId, String taskNumber,
|
||||
List<String> routeNodeIds,
|
||||
List<String> routeResourceKeys,
|
||||
long topologyVersion, long fencingToken,
|
||||
String expectedBootId, int currentNodeIndex,
|
||||
int releasedResourceCount,
|
||||
long lastDeviceSequence) {
|
||||
this.commandId = requireText(commandId, "commandId");
|
||||
this.commandKey = requireText(commandKey, "commandKey");
|
||||
this.deviceId = requireText(deviceId, "deviceId");
|
||||
this.taskNumber = requireText(taskNumber, "taskNumber");
|
||||
this.routeNodeIds = immutable(routeNodeIds, "routeNodeIds");
|
||||
this.routeResourceKeys = immutable(routeResourceKeys,
|
||||
"routeResourceKeys");
|
||||
if (this.routeResourceKeys.size() != this.routeNodeIds.size() * 2 - 1) {
|
||||
throw new IllegalArgumentException(
|
||||
"route resources must alternate node and edge resources");
|
||||
}
|
||||
if (topologyVersion <= 0L || fencingToken <= 0L) {
|
||||
throw new IllegalArgumentException("execution versions are invalid");
|
||||
}
|
||||
this.topologyVersion = topologyVersion;
|
||||
this.fencingToken = fencingToken;
|
||||
this.expectedBootId = requireText(expectedBootId, "expectedBootId");
|
||||
this.currentNodeIndex = currentNodeIndex;
|
||||
this.releasedResourceCount = releasedResourceCount;
|
||||
this.lastDeviceSequence = lastDeviceSequence;
|
||||
}
|
||||
|
||||
public Progress evaluateProgress(long deviceSequence, String currentNodeId,
|
||||
List<String> remainingRouteNodes,
|
||||
boolean completed) {
|
||||
if (deviceSequence <= lastDeviceSequence) {
|
||||
return Progress.stale(commandKey, currentNodeIndex,
|
||||
releasedResourceCount, lastDeviceSequence);
|
||||
}
|
||||
String current = requireText(currentNodeId, "currentNodeId");
|
||||
int index = routeNodeIds.indexOf(current);
|
||||
if (index < 0) {
|
||||
throw new DomainRuleViolation("PATH_MISMATCH");
|
||||
}
|
||||
if (index < currentNodeIndex) {
|
||||
throw new DomainRuleViolation("ROUTE_REGRESSION");
|
||||
}
|
||||
List<String> remaining = immutable(remainingRouteNodes,
|
||||
"remainingRouteNodes");
|
||||
if (!routeNodeIds.subList(index, routeNodeIds.size()).equals(remaining)) {
|
||||
throw new DomainRuleViolation("PATH_MISMATCH");
|
||||
}
|
||||
if (completed && index != routeNodeIds.size() - 1) {
|
||||
throw new DomainRuleViolation("COMPLETION_SITE_MISMATCH");
|
||||
}
|
||||
|
||||
int releaseUntil = completed ? routeResourceKeys.size() : index * 2;
|
||||
List<String> released = releaseUntil <= releasedResourceCount
|
||||
? Collections.<String>emptyList()
|
||||
: new ArrayList<String>(routeResourceKeys.subList(
|
||||
releasedResourceCount, releaseUntil));
|
||||
boolean advanced = index > currentNodeIndex || !released.isEmpty();
|
||||
return Progress.accepted(commandKey, currentNodeIndex,
|
||||
releasedResourceCount, lastDeviceSequence, index,
|
||||
Math.max(releasedResourceCount, releaseUntil), deviceSequence,
|
||||
released, advanced);
|
||||
}
|
||||
|
||||
public CommittedDeviceExecution advance(Progress progress) {
|
||||
if (progress == null || !commandKey.equals(progress.commandKey)) {
|
||||
throw new DomainRuleViolation("progress belongs to another command");
|
||||
}
|
||||
if (progress.stale) {
|
||||
return this;
|
||||
}
|
||||
if (currentNodeIndex != progress.baseNodeIndex
|
||||
|| releasedResourceCount != progress.baseReleasedResourceCount
|
||||
|| lastDeviceSequence != progress.baseDeviceSequence) {
|
||||
throw new DomainRuleViolation("execution progress version conflict");
|
||||
}
|
||||
return new CommittedDeviceExecution(commandId, commandKey, deviceId,
|
||||
taskNumber, routeNodeIds, routeResourceKeys, topologyVersion,
|
||||
fencingToken, expectedBootId, progress.nextNodeIndex,
|
||||
progress.nextReleasedResourceCount, progress.nextDeviceSequence);
|
||||
}
|
||||
|
||||
public CommittedDeviceExecution observeSequence(long deviceSequence) {
|
||||
if (deviceSequence <= lastDeviceSequence) {
|
||||
return this;
|
||||
}
|
||||
return new CommittedDeviceExecution(commandId, commandKey, deviceId,
|
||||
taskNumber, routeNodeIds, routeResourceKeys, topologyVersion,
|
||||
fencingToken, expectedBootId, currentNodeIndex,
|
||||
releasedResourceCount, deviceSequence);
|
||||
}
|
||||
|
||||
public String getCommandId() { return commandId; }
|
||||
public String getCommandKey() { return commandKey; }
|
||||
public String getDeviceId() { return deviceId; }
|
||||
public String getTaskNumber() { return taskNumber; }
|
||||
public List<String> getRouteNodeIds() { return routeNodeIds; }
|
||||
public List<String> getRouteResourceKeys() { return routeResourceKeys; }
|
||||
public long getTopologyVersion() { return topologyVersion; }
|
||||
public long getFencingToken() { return fencingToken; }
|
||||
public String getExpectedBootId() { return expectedBootId; }
|
||||
public long getLastDeviceSequence() { return lastDeviceSequence; }
|
||||
|
||||
public static final class Progress {
|
||||
private final String commandKey;
|
||||
private final boolean stale;
|
||||
private final boolean advanced;
|
||||
private final int baseNodeIndex;
|
||||
private final int baseReleasedResourceCount;
|
||||
private final long baseDeviceSequence;
|
||||
private final int nextNodeIndex;
|
||||
private final int nextReleasedResourceCount;
|
||||
private final long nextDeviceSequence;
|
||||
private final List<String> releasedResourceKeys;
|
||||
|
||||
private Progress(String commandKey, boolean stale, boolean advanced,
|
||||
int baseNodeIndex, int baseReleasedResourceCount,
|
||||
long baseDeviceSequence, int nextNodeIndex,
|
||||
int nextReleasedResourceCount, long nextDeviceSequence,
|
||||
List<String> releasedResourceKeys) {
|
||||
this.commandKey = commandKey;
|
||||
this.stale = stale;
|
||||
this.advanced = advanced;
|
||||
this.baseNodeIndex = baseNodeIndex;
|
||||
this.baseReleasedResourceCount = baseReleasedResourceCount;
|
||||
this.baseDeviceSequence = baseDeviceSequence;
|
||||
this.nextNodeIndex = nextNodeIndex;
|
||||
this.nextReleasedResourceCount = nextReleasedResourceCount;
|
||||
this.nextDeviceSequence = nextDeviceSequence;
|
||||
this.releasedResourceKeys = Collections.unmodifiableList(
|
||||
new ArrayList<String>(releasedResourceKeys));
|
||||
}
|
||||
|
||||
static Progress stale(String commandKey, int nodeIndex,
|
||||
int releasedResourceCount, long sequence) {
|
||||
return new Progress(commandKey, true, false, nodeIndex,
|
||||
releasedResourceCount, sequence, nodeIndex,
|
||||
releasedResourceCount, sequence,
|
||||
Collections.<String>emptyList());
|
||||
}
|
||||
|
||||
static Progress accepted(String commandKey, int baseNodeIndex,
|
||||
int baseReleasedResourceCount,
|
||||
long baseDeviceSequence, int nextNodeIndex,
|
||||
int nextReleasedResourceCount,
|
||||
long nextDeviceSequence, List<String> released,
|
||||
boolean advanced) {
|
||||
return new Progress(commandKey, false, advanced, baseNodeIndex,
|
||||
baseReleasedResourceCount, baseDeviceSequence, nextNodeIndex,
|
||||
nextReleasedResourceCount, nextDeviceSequence, released);
|
||||
}
|
||||
|
||||
public boolean isStale() { return stale; }
|
||||
public boolean isAdvanced() { return advanced; }
|
||||
public List<String> getReleasedResourceKeys() {
|
||||
return releasedResourceKeys;
|
||||
}
|
||||
}
|
||||
|
||||
private static List<String> immutable(List<String> source, String field) {
|
||||
if (source == null || source.isEmpty()) {
|
||||
throw new IllegalArgumentException(field + " must not be empty");
|
||||
}
|
||||
List<String> copy = new ArrayList<String>();
|
||||
for (String value : source) {
|
||||
copy.add(requireText(value, field));
|
||||
}
|
||||
return Collections.unmodifiableList(copy);
|
||||
}
|
||||
|
||||
private static String requireText(String value, String field) {
|
||||
if (value == null || value.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException(field + " must not be blank");
|
||||
}
|
||||
return value.trim();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package com.ruoyi.wcs.domain.execution;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
public final class DeviceEvent {
|
||||
public enum Type {
|
||||
ACKNOWLEDGED,
|
||||
PROGRESS,
|
||||
COMPLETED,
|
||||
REJECTED,
|
||||
FAULT
|
||||
}
|
||||
|
||||
private final String eventId;
|
||||
private final String deviceId;
|
||||
private final String commandKey;
|
||||
private final long deviceSequence;
|
||||
private final Instant deviceTime;
|
||||
private final Instant receivedTime;
|
||||
private final Type eventType;
|
||||
private final String taskNumber;
|
||||
private final String currentNodeId;
|
||||
private final List<String> remainingRouteNodes;
|
||||
|
||||
public DeviceEvent(String eventId, String deviceId, String commandKey,
|
||||
long deviceSequence, Instant deviceTime,
|
||||
Instant receivedTime, Type eventType, String taskNumber,
|
||||
String currentNodeId, List<String> remainingRouteNodes) {
|
||||
this.eventId = requireText(eventId, "eventId");
|
||||
this.deviceId = requireText(deviceId, "deviceId");
|
||||
this.commandKey = requireText(commandKey, "commandKey");
|
||||
if (deviceSequence < 0L) {
|
||||
throw new IllegalArgumentException("deviceSequence must not be negative");
|
||||
}
|
||||
this.deviceSequence = deviceSequence;
|
||||
this.deviceTime = Objects.requireNonNull(deviceTime, "deviceTime");
|
||||
this.receivedTime = Objects.requireNonNull(receivedTime, "receivedTime");
|
||||
this.eventType = Objects.requireNonNull(eventType, "eventType");
|
||||
this.taskNumber = requireText(taskNumber, "taskNumber");
|
||||
boolean routeFactsRequired = eventType == Type.PROGRESS
|
||||
|| eventType == Type.COMPLETED;
|
||||
if (routeFactsRequired) {
|
||||
this.currentNodeId = requireText(currentNodeId, "currentNodeId");
|
||||
this.remainingRouteNodes = immutable(remainingRouteNodes,
|
||||
"remainingRouteNodes");
|
||||
} else {
|
||||
this.currentNodeId = currentNodeId == null ? null
|
||||
: requireText(currentNodeId, "currentNodeId");
|
||||
this.remainingRouteNodes = remainingRouteNodes == null
|
||||
? Collections.<String>emptyList()
|
||||
: immutable(remainingRouteNodes, "remainingRouteNodes");
|
||||
}
|
||||
}
|
||||
|
||||
public String getEventId() {
|
||||
return eventId;
|
||||
}
|
||||
|
||||
public String getDeviceId() {
|
||||
return deviceId;
|
||||
}
|
||||
|
||||
public String getCommandKey() {
|
||||
return commandKey;
|
||||
}
|
||||
|
||||
public long getDeviceSequence() {
|
||||
return deviceSequence;
|
||||
}
|
||||
|
||||
public Instant getDeviceTime() {
|
||||
return deviceTime;
|
||||
}
|
||||
|
||||
public Instant getReceivedTime() {
|
||||
return receivedTime;
|
||||
}
|
||||
|
||||
public Type getEventType() {
|
||||
return eventType;
|
||||
}
|
||||
|
||||
public String getTaskNumber() {
|
||||
return taskNumber;
|
||||
}
|
||||
|
||||
public String getCurrentNodeId() {
|
||||
return currentNodeId;
|
||||
}
|
||||
|
||||
public List<String> getRemainingRouteNodes() {
|
||||
return remainingRouteNodes;
|
||||
}
|
||||
|
||||
private static List<String> immutable(List<String> source, String field) {
|
||||
if (source == null || source.isEmpty()) {
|
||||
throw new IllegalArgumentException(field + " must not be empty");
|
||||
}
|
||||
List<String> copy = new ArrayList<String>();
|
||||
for (String value : source) {
|
||||
copy.add(requireText(value, field));
|
||||
}
|
||||
return Collections.unmodifiableList(copy);
|
||||
}
|
||||
|
||||
private static String requireText(String value, String field) {
|
||||
if (value == null || value.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException(field + " must not be blank");
|
||||
}
|
||||
return value.trim();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package com.ruoyi.wcs.domain.execution;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
public final class DeviceSnapshot {
|
||||
private final String snapshotId;
|
||||
private final String deviceId;
|
||||
private final String commandKey;
|
||||
private final String taskNumber;
|
||||
private final long deviceSequence;
|
||||
private final Instant observedAt;
|
||||
private final boolean connected;
|
||||
private final String deviceBootId;
|
||||
private final long topologyVersion;
|
||||
private final long fencingToken;
|
||||
private final String currentNodeId;
|
||||
private final List<String> remainingRouteNodes;
|
||||
private final boolean commandCompleted;
|
||||
|
||||
public DeviceSnapshot(String snapshotId, String deviceId, String commandKey,
|
||||
String taskNumber, long deviceSequence,
|
||||
Instant observedAt, boolean connected,
|
||||
String deviceBootId, long topologyVersion,
|
||||
long fencingToken, String currentNodeId,
|
||||
List<String> remainingRouteNodes,
|
||||
boolean commandCompleted) {
|
||||
this.snapshotId = requireText(snapshotId, "snapshotId");
|
||||
this.deviceId = requireText(deviceId, "deviceId");
|
||||
this.commandKey = requireText(commandKey, "commandKey");
|
||||
this.taskNumber = requireText(taskNumber, "taskNumber");
|
||||
if (deviceSequence < 0L || topologyVersion <= 0L || fencingToken <= 0L) {
|
||||
throw new IllegalArgumentException("snapshot versions are invalid");
|
||||
}
|
||||
this.deviceSequence = deviceSequence;
|
||||
this.observedAt = Objects.requireNonNull(observedAt, "observedAt");
|
||||
this.connected = connected;
|
||||
this.deviceBootId = requireText(deviceBootId, "deviceBootId");
|
||||
this.topologyVersion = topologyVersion;
|
||||
this.fencingToken = fencingToken;
|
||||
this.currentNodeId = requireText(currentNodeId, "currentNodeId");
|
||||
this.remainingRouteNodes = immutable(remainingRouteNodes);
|
||||
this.commandCompleted = commandCompleted;
|
||||
}
|
||||
|
||||
public String getSnapshotId() { return snapshotId; }
|
||||
public String getDeviceId() { return deviceId; }
|
||||
public String getCommandKey() { return commandKey; }
|
||||
public String getTaskNumber() { return taskNumber; }
|
||||
public long getDeviceSequence() { return deviceSequence; }
|
||||
public Instant getObservedAt() { return observedAt; }
|
||||
public boolean isConnected() { return connected; }
|
||||
public String getDeviceBootId() { return deviceBootId; }
|
||||
public long getTopologyVersion() { return topologyVersion; }
|
||||
public long getFencingToken() { return fencingToken; }
|
||||
public String getCurrentNodeId() { return currentNodeId; }
|
||||
public List<String> getRemainingRouteNodes() { return remainingRouteNodes; }
|
||||
public boolean isCommandCompleted() { return commandCompleted; }
|
||||
|
||||
private static List<String> immutable(List<String> source) {
|
||||
if (source == null || source.isEmpty()) {
|
||||
throw new IllegalArgumentException("remainingRouteNodes must not be empty");
|
||||
}
|
||||
List<String> copy = new ArrayList<String>();
|
||||
for (String value : source) {
|
||||
copy.add(requireText(value, "remainingRouteNodes"));
|
||||
}
|
||||
return Collections.unmodifiableList(copy);
|
||||
}
|
||||
|
||||
private static String requireText(String value, String field) {
|
||||
if (value == null || value.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException(field + " must not be blank");
|
||||
}
|
||||
return value.trim();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ruoyi.wcs.port.in;
|
||||
|
||||
import com.ruoyi.wcs.application.device.EventAcceptanceResult;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceEvent;
|
||||
|
||||
public interface DeviceEventUseCase {
|
||||
EventAcceptanceResult accept(DeviceEvent event);
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ruoyi.wcs.port.out;
|
||||
|
||||
import com.ruoyi.wcs.application.device.DeviceCommandEnvelope;
|
||||
import com.ruoyi.wcs.application.device.SendReceipt;
|
||||
|
||||
public interface DeviceCommandGateway {
|
||||
SendReceipt send(DeviceCommandEnvelope envelope);
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ruoyi.wcs.port.out;
|
||||
|
||||
import com.ruoyi.wcs.domain.execution.DeviceEvent;
|
||||
|
||||
public interface DeviceEventRepository {
|
||||
boolean claim(DeviceEvent event);
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package com.ruoyi.wcs.port.out;
|
||||
|
||||
import com.ruoyi.wcs.domain.execution.CommittedDeviceExecution;
|
||||
|
||||
public interface DeviceExecutionRepository {
|
||||
CommittedDeviceExecution findByCommandKey(String commandKey);
|
||||
|
||||
CommittedDeviceExecution findActiveByDeviceId(String deviceId);
|
||||
|
||||
void save(CommittedDeviceExecution execution);
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ruoyi.wcs.port.out;
|
||||
|
||||
import com.ruoyi.wcs.domain.execution.DeviceSnapshot;
|
||||
|
||||
public interface DeviceSnapshotGateway {
|
||||
DeviceSnapshot readSnapshot(String deviceId);
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ruoyi.wcs.port.out;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public interface RecoveryCaseOpener {
|
||||
void open(String deviceId, String commandKey, String reason,
|
||||
List<String> evidenceIds);
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package com.ruoyi.wcs.port.out;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public interface ResourceReservationRepository {
|
||||
void releasePassedResources(String commandKey, List<String> resourceKeys,
|
||||
String releaseEvidenceId, long fencingToken);
|
||||
|
||||
void protectForUnknown(String commandKey, String evidenceId);
|
||||
}
|
||||
@@ -0,0 +1,293 @@
|
||||
package com.ruoyi.wcs.application.device;
|
||||
|
||||
import com.ruoyi.wcs.domain.execution.CommittedDeviceExecution;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceCommand;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceCommandState;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceCommandType;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceEvent;
|
||||
import com.ruoyi.wcs.domain.execution.DeviceSnapshot;
|
||||
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.RecoveryCaseOpener;
|
||||
import com.ruoyi.wcs.port.out.ResourceReservationRepository;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
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.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class FeedbackReconcilerTest {
|
||||
private FakeCommandRepository commands;
|
||||
private FakeExecutionRepository executions;
|
||||
private FakeEventRepository events;
|
||||
private FakeReservationRepository reservations;
|
||||
private FakeRecoveryCaseOpener recovery;
|
||||
private FeedbackReconciler reconciler;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
commands = new FakeCommandRepository();
|
||||
executions = new FakeExecutionRepository();
|
||||
events = new FakeEventRepository();
|
||||
reservations = new FakeReservationRepository();
|
||||
recovery = new FakeRecoveryCaseOpener();
|
||||
reconciler = new FeedbackReconciler(commands, executions, events,
|
||||
reservations, recovery);
|
||||
|
||||
DeviceCommand command = DeviceCommand.pending(
|
||||
"CMD-1", "CMD-KEY-1", "VEH-1", DeviceCommandType.MOVE_ROUTE,
|
||||
"{}", 1, at(0));
|
||||
command.markSent("SEND-1", at(1));
|
||||
commands.put(command);
|
||||
executions.put(execution());
|
||||
commands.resetUpdateCount();
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateEventDoesNotAdvanceTheCommandTwice() {
|
||||
DeviceEvent completed = new DeviceEvent(
|
||||
"E-1", "VEH-1", "CMD-KEY-1", 10L, at(10), at(11),
|
||||
DeviceEvent.Type.COMPLETED, "TASK-1", "C",
|
||||
Collections.singletonList("C"));
|
||||
|
||||
EventAcceptanceResult first = reconciler.accept(completed);
|
||||
EventAcceptanceResult duplicate = reconciler.accept(completed);
|
||||
|
||||
assertEquals(EventAcceptanceResult.APPLIED, first);
|
||||
assertEquals(EventAcceptanceResult.DUPLICATE, duplicate);
|
||||
assertEquals(1, commands.updateCount("CMD-1"));
|
||||
assertEquals(1, reservations.releaseCallCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
void taskNumberMismatchCreatesRecoveryInsteadOfReleasing() {
|
||||
DeviceSnapshot snapshot = snapshot("WRONG-TASK", "B",
|
||||
Arrays.asList("B", "C"), 5L, false);
|
||||
|
||||
ReconcileResult result = reconciler.reconcile(snapshot);
|
||||
|
||||
assertEquals(ReconcileStatus.RECOVERY_REQUIRED, result.getStatus());
|
||||
assertEquals(0, reservations.releaseCallCount);
|
||||
assertEquals(Collections.singletonList("TASK_NUMBER_MISMATCH"),
|
||||
recovery.reasons);
|
||||
}
|
||||
|
||||
@Test
|
||||
void releasesOnlyResourcesStrictlyPassedByMatchingProgress() {
|
||||
DeviceSnapshot snapshot = snapshot("TASK-1", "B",
|
||||
Arrays.asList("B", "C"), 5L, false);
|
||||
|
||||
ReconcileResult result = reconciler.reconcile(snapshot);
|
||||
|
||||
assertEquals(ReconcileStatus.ADVANCED, result.getStatus());
|
||||
assertEquals(Arrays.asList("NODE:A", "EDGE:A:B"),
|
||||
result.getReleasedResourceKeys());
|
||||
assertEquals(Arrays.asList("NODE:A", "EDGE:A:B"),
|
||||
reservations.releasedResources);
|
||||
}
|
||||
|
||||
@Test
|
||||
void staleSnapshotIsIgnoredWithoutReleasingAgain() {
|
||||
reconciler.reconcile(snapshot("TASK-1", "B",
|
||||
Arrays.asList("B", "C"), 5L, false));
|
||||
|
||||
ReconcileResult stale = reconciler.reconcile(snapshot("TASK-1", "C",
|
||||
Collections.singletonList("C"), 4L, false));
|
||||
|
||||
assertEquals(ReconcileStatus.IGNORED_STALE, stale.getStatus());
|
||||
assertEquals(1, reservations.releaseCallCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
void unknownCommandUsesNonTerminalSnapshotForDiagnosticsOnly() {
|
||||
DeviceCommand command = commands.findByCommandKey("CMD-KEY-1");
|
||||
command.markUnknown("WRITE_RESULT_AMBIGUOUS", at(2));
|
||||
commands.save(command);
|
||||
commands.resetUpdateCount();
|
||||
|
||||
ReconcileResult result = reconciler.reconcile(snapshot("TASK-1", "B",
|
||||
Arrays.asList("B", "C"), 5L, false));
|
||||
|
||||
assertEquals(ReconcileStatus.CONSISTENT, result.getStatus());
|
||||
assertTrue(result.getReleasedResourceKeys().isEmpty());
|
||||
assertEquals(0, reservations.releaseCallCount);
|
||||
assertEquals(-1L, executions.findByCommandKey("CMD-KEY-1")
|
||||
.getLastDeviceSequence());
|
||||
assertEquals(DeviceCommandState.UNKNOWN, command.getState());
|
||||
}
|
||||
|
||||
@Test
|
||||
void terminalCommandCannotReleaseResourcesFromLateCompletion() {
|
||||
DeviceCommand command = commands.findByCommandKey("CMD-KEY-1");
|
||||
command.reject("PLC_REJECTED", at(2));
|
||||
commands.save(command);
|
||||
commands.resetUpdateCount();
|
||||
|
||||
ReconcileResult result = reconciler.reconcile(snapshot("TASK-1", "C",
|
||||
Collections.singletonList("C"), 5L, true));
|
||||
|
||||
assertEquals(ReconcileStatus.RECOVERY_REQUIRED, result.getStatus());
|
||||
assertEquals(0, reservations.releaseCallCount);
|
||||
assertEquals(-1L, executions.findByCommandKey("CMD-KEY-1")
|
||||
.getLastDeviceSequence());
|
||||
}
|
||||
|
||||
@Test
|
||||
void acknowledgementDoesNotRequireRouteFacts() {
|
||||
DeviceEvent acknowledgement = new DeviceEvent(
|
||||
"ACK-1", "VEH-1", "CMD-KEY-1", 2L, at(2), at(2),
|
||||
DeviceEvent.Type.ACKNOWLEDGED, "TASK-1", null, null);
|
||||
|
||||
assertEquals(EventAcceptanceResult.APPLIED,
|
||||
reconciler.accept(acknowledgement));
|
||||
assertEquals(DeviceCommandState.ACKNOWLEDGED,
|
||||
commands.findByCommandKey("CMD-KEY-1").getState());
|
||||
assertEquals(0, reservations.releaseCallCount);
|
||||
}
|
||||
|
||||
private static CommittedDeviceExecution execution() {
|
||||
return new CommittedDeviceExecution(
|
||||
"CMD-1", "CMD-KEY-1", "VEH-1", "TASK-1",
|
||||
Arrays.asList("A", "B", "C"),
|
||||
Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B",
|
||||
"EDGE:B:C", "NODE:C"),
|
||||
7L, 11L, "BOOT-1");
|
||||
}
|
||||
|
||||
private static DeviceSnapshot snapshot(String taskNumber, String currentNode,
|
||||
List<String> remaining, long sequence,
|
||||
boolean completed) {
|
||||
return new DeviceSnapshot("SNAP-" + sequence, "VEH-1",
|
||||
"CMD-KEY-1", taskNumber, sequence, at(sequence), true,
|
||||
"BOOT-1", 7L, 11L, currentNode, remaining, completed);
|
||||
}
|
||||
|
||||
private static Instant at(long seconds) {
|
||||
return Instant.EPOCH.plusSeconds(seconds);
|
||||
}
|
||||
|
||||
private static final class FakeCommandRepository
|
||||
implements DeviceCommandRepository {
|
||||
private final Map<String, DeviceCommand> byId =
|
||||
new HashMap<String, DeviceCommand>();
|
||||
private final Map<String, Integer> updates =
|
||||
new HashMap<String, Integer>();
|
||||
|
||||
void put(DeviceCommand command) {
|
||||
byId.put(command.getCommandId(), command);
|
||||
}
|
||||
|
||||
void resetUpdateCount() {
|
||||
updates.clear();
|
||||
}
|
||||
|
||||
int updateCount(String commandId) {
|
||||
Integer count = updates.get(commandId);
|
||||
return count == null ? 0 : count;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DeviceCommand findByCommandKey(String commandKey) {
|
||||
for (DeviceCommand command : byId.values()) {
|
||||
if (command.getCommandKey().equals(commandKey)) {
|
||||
return command;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DeviceCommand findByCommandId(String commandId) {
|
||||
return byId.get(commandId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void save(DeviceCommand command) {
|
||||
byId.put(command.getCommandId(), command);
|
||||
Integer count = updates.get(command.getCommandId());
|
||||
updates.put(command.getCommandId(), count == null ? 1 : count + 1);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class FakeExecutionRepository
|
||||
implements DeviceExecutionRepository {
|
||||
private final Map<String, CommittedDeviceExecution> byCommand =
|
||||
new HashMap<String, CommittedDeviceExecution>();
|
||||
|
||||
void put(CommittedDeviceExecution execution) {
|
||||
byCommand.put(execution.getCommandKey(), execution);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CommittedDeviceExecution findByCommandKey(String commandKey) {
|
||||
return byCommand.get(commandKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CommittedDeviceExecution findActiveByDeviceId(String deviceId) {
|
||||
for (CommittedDeviceExecution execution : byCommand.values()) {
|
||||
if (execution.getDeviceId().equals(deviceId)) {
|
||||
return execution;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void save(CommittedDeviceExecution execution) {
|
||||
put(execution);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class FakeEventRepository
|
||||
implements DeviceEventRepository {
|
||||
private final Set<String> ids = new HashSet<String>();
|
||||
|
||||
@Override
|
||||
public boolean claim(DeviceEvent event) {
|
||||
return ids.add(event.getEventId());
|
||||
}
|
||||
}
|
||||
|
||||
private static final class FakeReservationRepository
|
||||
implements ResourceReservationRepository {
|
||||
private int releaseCallCount;
|
||||
private List<String> releasedResources = Collections.emptyList();
|
||||
|
||||
@Override
|
||||
public void releasePassedResources(String commandKey,
|
||||
List<String> resourceKeys,
|
||||
String releaseEvidenceId,
|
||||
long fencingToken) {
|
||||
releaseCallCount++;
|
||||
releasedResources = new ArrayList<String>(resourceKeys);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void protectForUnknown(String commandKey, String evidenceId) {
|
||||
}
|
||||
}
|
||||
|
||||
private static final class FakeRecoveryCaseOpener
|
||||
implements RecoveryCaseOpener {
|
||||
private final List<String> reasons = new ArrayList<String>();
|
||||
|
||||
@Override
|
||||
public void open(String deviceId, String commandKey, String reason,
|
||||
List<String> evidenceIds) {
|
||||
reasons.add(reason);
|
||||
assertTrue(!evidenceIds.isEmpty());
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user