lhq/FT546-WCS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ft546): serialize equipment command sends

Browse Source
This commit is contained in:
zgw
2026-08-11 14:39:26 +08:00
parent 14a5a6174e
commit e5f91de1a0
4 changed files with 213 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/AbstractFt546EquipmentAdapter.java
View File

@@ -15,6 +15,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
public abstract class AbstractFt546EquipmentAdapter
implements DeviceCommandGateway, DeviceSnapshotGateway {
@@ -25,6 +26,7 @@ public abstract class AbstractFt546EquipmentAdapter
private final Set<String> locallyUnknownDevices =
Collections.newSetFromMap(
new ConcurrentHashMap<String, Boolean>());
private final ReentrantLock sendLock = new ReentrantLock();
protected AbstractFt546EquipmentAdapter(
Ft546DeviceProfile sourceProfile, String expectedDeviceType,
@@ -62,32 +64,41 @@ public abstract class AbstractFt546EquipmentAdapter
return SendReceipt.rejected(receiptId,
"command type is not supported by this adapter");
}
if (locallyUnknownDevices.contains(envelope.getDeviceId())
|| motionCommandGuard.hasBlockingMotionCommand(
envelope.getDeviceId(), envelope.getCommandId())) {
if (!sendLock.tryLock()) {
return SendReceipt.rejected(receiptId,
"device has another active command or unresolved UNKNOWN");
}
final byte[] payload;
try {
payload = encode(envelope.getPayload());
} catch (IllegalArgumentException rejected) {
return SendReceipt.rejected(receiptId, rejected.getMessage());
}
if (!isConnected()) {
return SendReceipt.notSent("equipment PLC is disconnected");
"device command send is already in progress");
}
try {
int result = write(commandAddress(), payload);
if (result <= 0) {
return markUnknown(envelope.getDeviceId(), receiptId,
"equipment write result is ambiguous");
if (locallyUnknownDevices.contains(envelope.getDeviceId())
|| motionCommandGuard.hasBlockingMotionCommand(
envelope.getDeviceId(), envelope.getCommandId())) {
return SendReceipt.rejected(receiptId,
"device has another active command or unresolved UNKNOWN");
}
return SendReceipt.accepted(receiptId);
} catch (RuntimeException ambiguous) {
return markUnknown(envelope.getDeviceId(), receiptId,
"equipment write raised "
+ ambiguous.getClass().getSimpleName());
final byte[] payload;
try {
payload = encode(envelope.getPayload());
} catch (IllegalArgumentException rejected) {
return SendReceipt.rejected(receiptId,
rejected.getMessage());
}
if (!isConnected()) {
return SendReceipt.notSent("equipment PLC is disconnected");
}
try {
int result = write(commandAddress(), payload);
if (result <= 0) {
return markUnknown(envelope.getDeviceId(), receiptId,
"equipment write result is ambiguous");
}
return SendReceipt.accepted(receiptId);
} catch (RuntimeException ambiguous) {
return markUnknown(envelope.getDeviceId(), receiptId,
"equipment write raised "
+ ambiguous.getClass().getSimpleName());
}
} finally {
sendLock.unlock();
}
}

87
ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapter.java
View File

@@ -17,6 +17,7 @@ import java.util.Collections;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
public final class Ft546LegacySrgvAdapter
implements DeviceCommandGateway, DeviceSnapshotGateway {
@@ -32,6 +33,7 @@ public final class Ft546LegacySrgvAdapter
private final Set<String> locallyUnknownDevices =
Collections.newSetFromMap(
new ConcurrentHashMap<String, Boolean>());
private final ReentrantLock sendLock = new ReentrantLock();
public Ft546LegacySrgvAdapter(Ft546DeviceRuntimeRegistry registry,
Ft546DeviceProfile profile,
@@ -66,49 +68,58 @@ public final class Ft546LegacySrgvAdapter
return SendReceipt.rejected(receiptId,
"profile belongs to another device");
}
if (locallyUnknownDevices.contains(envelope.getDeviceId())
|| motionCommandGuard.hasBlockingMotionCommand(
envelope.getDeviceId(), envelope.getCommandId())) {
if (!sendLock.tryLock()) {
return SendReceipt.rejected(receiptId,
"device has another active motion command or an "
+ "unresolved UNKNOWN command");
"device command send is already in progress");
}
final Ft546SrgvCommandEncoder.EncodedCommand command;
try {
command = encoder.encode(envelope, profile, capabilities);
} catch (IllegalArgumentException | IllegalStateException rejected) {
return SendReceipt.rejected(receiptId, rejected.getMessage());
}
final PlcSrgvRead runtime;
try {
runtime = registry.requireSrgv(envelope.getDeviceId());
} catch (IllegalArgumentException missingRuntime) {
return SendReceipt.notSent(missingRuntime.getMessage());
}
if (!runtime.IsConnect()) {
return SendReceipt.notSent("shuttle PLC is disconnected");
}
boolean wroteAny = false;
for (Ft546SrgvCommandEncoder.PlcWrite write
: command.getWrites()) {
try {
if (!write.writeTo(runtime)) {
return markUnknown(envelope.getDeviceId(), receiptId,
wroteAny
? "PLC rejected a partial register sequence"
: "first register write result is ambiguous");
}
wroteAny = true;
} catch (RuntimeException ambiguous) {
return markUnknown(envelope.getDeviceId(), receiptId,
"register write result is ambiguous: "
+ ambiguous.getClass().getSimpleName());
if (locallyUnknownDevices.contains(envelope.getDeviceId())
|| motionCommandGuard.hasBlockingMotionCommand(
envelope.getDeviceId(), envelope.getCommandId())) {
return SendReceipt.rejected(receiptId,
"device has another active motion command or an "
+ "unresolved UNKNOWN command");
}
final Ft546SrgvCommandEncoder.EncodedCommand command;
try {
command = encoder.encode(envelope, profile, capabilities);
} catch (IllegalArgumentException | IllegalStateException rejected) {
return SendReceipt.rejected(receiptId,
rejected.getMessage());
}
final PlcSrgvRead runtime;
try {
runtime = registry.requireSrgv(envelope.getDeviceId());
} catch (IllegalArgumentException missingRuntime) {
return SendReceipt.notSent(missingRuntime.getMessage());
}
if (!runtime.IsConnect()) {
return SendReceipt.notSent("shuttle PLC is disconnected");
}
boolean wroteAny = false;
for (Ft546SrgvCommandEncoder.PlcWrite write
: command.getWrites()) {
try {
if (!write.writeTo(runtime)) {
return markUnknown(envelope.getDeviceId(), receiptId,
wroteAny
? "PLC rejected a partial register sequence"
: "first register write result is ambiguous");
}
wroteAny = true;
} catch (RuntimeException ambiguous) {
return markUnknown(envelope.getDeviceId(), receiptId,
"register write result is ambiguous: "
+ ambiguous.getClass().getSimpleName());
}
}
return SendReceipt.accepted(receiptId);
} finally {
sendLock.unlock();
}
return SendReceipt.accepted(receiptId);
}
@Override

66
ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/Ft546EquipmentAdaptersTest.java
View File

@@ -24,10 +24,17 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class Ft546EquipmentAdaptersTest {
private static final Instant NOW = Instant.parse("2026-08-11T05:00:00Z");
@@ -153,6 +160,35 @@ class Ft546EquipmentAdaptersTest {
assertEquals(1, runtime.payloads.size());
}
@Test
void rejectsConcurrentEquipmentSendWhileFirstWriteIsInProgress()
throws Exception {
FakeCdzRead runtime = new FakeCdzRead("1", chargerDevice());
runtime.blockFirstWrite();
Ft546ChargerAdapter adapter = new Ft546ChargerAdapter(runtime,
profile("1", "CHARGER", "CHARGE", "DB701.0"),
(deviceId, commandId) -> false, contextProvider());
ExecutorService executor = Executors.newSingleThreadExecutor();
try {
Future<SendReceipt> first = executor.submit(() -> adapter.send(
envelope("CHARGE-1", "1", DeviceCommandType.CHARGE,
"enabled=1")));
assertTrue(runtime.firstWriteEntered.await(2, TimeUnit.SECONDS));
SendReceipt second = adapter.send(envelope("CHARGE-2", "1",
DeviceCommandType.CHARGE, "enabled=1"));
runtime.releaseFirstWrite.countDown();
assertEquals(SendReceipt.Status.ACCEPTED,
first.get(2, TimeUnit.SECONDS).getStatus());
assertEquals(SendReceipt.Status.REJECTED, second.getStatus());
assertEquals(1, runtime.writeCalls.get());
} finally {
runtime.releaseFirstWrite.countDown();
executor.shutdownNow();
}
}
@Test
void factReadersPublishAConnectionDropAfterTheyWereConnected() {
FakeTsjRead lift = new FakeTsjRead("1", liftDevice());
@@ -266,9 +302,17 @@ class Ft546EquipmentAdaptersTest {
}
private static final class FakeCdzRead extends PlcCdzRead {
private final List<String> addresses = new ArrayList<String>();
private final List<byte[]> payloads = new ArrayList<byte[]>();
private final List<String> addresses = Collections.synchronizedList(
new ArrayList<String>());
private final List<byte[]> payloads = Collections.synchronizedList(
new ArrayList<byte[]>());
private final CountDownLatch firstWriteEntered =
new CountDownLatch(1);
private final CountDownLatch releaseFirstWrite =
new CountDownLatch(1);
private int writeResult = 1;
private volatile boolean blockFirstWrite;
private final AtomicInteger writeCalls = new AtomicInteger();
private FakeCdzRead(String id, DeviceCdz device) {
super(map(id, device), new PlcConnectInfo("127.0.0.1", 102));
@@ -277,10 +321,28 @@ class Ft546EquipmentAdaptersTest {
@Override
public int write(String address, byte[] payload) {
int call = writeCalls.incrementAndGet();
if (blockFirstWrite && call == 1) {
firstWriteEntered.countDown();
await(releaseFirstWrite);
}
addresses.add(address);
payloads.add(payload.clone());
return writeResult;
}
private void blockFirstWrite() {
blockFirstWrite = true;
}
private static void await(CountDownLatch latch) {
try {
latch.await();
} catch (InterruptedException interrupted) {
Thread.currentThread().interrupt();
throw new IllegalStateException(interrupted);
}
}
}
private static ConcurrentHashMap<String, DeviceTsj> map(

70
ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapterTest.java
View File

@@ -22,6 +22,12 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -149,6 +155,34 @@ class Ft546LegacySrgvAdapterTest {
assertTrue(runtime.writes.isEmpty());
}
@Test
void rejectsConcurrentShuttleSendWithoutInterleavingRegisterSequences()
throws Exception {
FakePlcSrgvRead runtime = connectedRuntime("1");
runtime.blockFirstWrite();
Ft546LegacySrgvAdapter adapter = adapter(runtime, pointProfile(),
null, (deviceId, commandId) -> false);
ExecutorService executor = Executors.newSingleThreadExecutor();
try {
Future<SendReceipt> first = executor.submit(() -> adapter.send(
envelope("CMD-1",
"taskNo=30001|taskType=1|route=070501>070502")));
assertTrue(runtime.firstWriteEntered.await(2, TimeUnit.SECONDS));
SendReceipt second = adapter.send(envelope("CMD-2",
"taskNo=30002|taskType=1|route=070503>070504"));
runtime.releaseFirstWrite.countDown();
assertEquals(SendReceipt.Status.ACCEPTED,
first.get(2, TimeUnit.SECONDS).getStatus());
assertEquals(SendReceipt.Status.REJECTED, second.getStatus());
assertEquals(4, runtime.writes.size());
} finally {
runtime.releaseFirstWrite.countDown();
executor.shutdownNow();
}
}
@Test
void twoStageProfileIsRejectedUntilExactEvidencePasses() {
FakePlcSrgvRead runtime = connectedRuntime("1");
@@ -340,23 +374,32 @@ class Ft546LegacySrgvAdapterTest {
private static final class FakePlcSrgvRead extends PlcSrgvRead {
private final List<RecordedWrite> writes =
new ArrayList<RecordedWrite>();
Collections.synchronizedList(
new ArrayList<RecordedWrite>());
private final CountDownLatch firstWriteEntered =
new CountDownLatch(1);
private final CountDownLatch releaseFirstWrite =
new CountDownLatch(1);
private String throwAfterAddress;
private String returnFalseAfterAddress;
private volatile boolean blockFirstWrite;
private final AtomicInteger writeCalls = new AtomicInteger();
private FakePlcSrgvRead(DeviceSrgv device) {
super(device, new PlcConnectInfo("127.0.0.1", 502));
}
@Override
public synchronized boolean Write(String address, int[] value) {
public boolean Write(String address, int[] value) {
beforeWrite();
record(address, "INT32", value);
maybeThrow(address);
return !address.equals(returnFalseAfterAddress);
}
@Override
public synchronized boolean Write(String address, short[] value) {
public boolean Write(String address, short[] value) {
beforeWrite();
List<Integer> values = new ArrayList<Integer>();
for (short item : value) {
values.add((int) item);
@@ -366,6 +409,27 @@ class Ft546LegacySrgvAdapterTest {
return !address.equals(returnFalseAfterAddress);
}
private void blockFirstWrite() {
blockFirstWrite = true;
}
private void beforeWrite() {
int call = writeCalls.incrementAndGet();
if (blockFirstWrite && call == 1) {
firstWriteEntered.countDown();
await(releaseFirstWrite);
}
}
private static void await(CountDownLatch latch) {
try {
latch.await();
} catch (InterruptedException interrupted) {
Thread.currentThread().interrupt();
throw new IllegalStateException(interrupted);
}
}
private void record(String address, String kind, int[] values) {
List<Integer> converted = new ArrayList<Integer>();
Arrays.stream(values).forEach(converted::add);