diff --git a/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/AbstractFt546EquipmentAdapter.java b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/AbstractFt546EquipmentAdapter.java index 7ecea7f..f5680d6 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/AbstractFt546EquipmentAdapter.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/AbstractFt546EquipmentAdapter.java @@ -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 locallyUnknownDevices = Collections.newSetFromMap( new ConcurrentHashMap()); + 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(); } } diff --git a/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapter.java b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapter.java index 5a0b46d..95789c2 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapter.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapter.java @@ -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 locallyUnknownDevices = Collections.newSetFromMap( new ConcurrentHashMap()); + 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 diff --git a/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/Ft546EquipmentAdaptersTest.java b/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/Ft546EquipmentAdaptersTest.java index 386a452..e1a267d 100644 --- a/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/Ft546EquipmentAdaptersTest.java +++ b/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/Ft546EquipmentAdaptersTest.java @@ -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 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 addresses = new ArrayList(); - private final List payloads = new ArrayList(); + private final List addresses = Collections.synchronizedList( + new ArrayList()); + private final List payloads = Collections.synchronizedList( + new ArrayList()); + 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 map( diff --git a/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapterTest.java b/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapterTest.java index 0365219..fc113f4 100644 --- a/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapterTest.java +++ b/ruoyi-admin/src/test/java/com/ruoyi/project/ft546/device/shuttle/Ft546LegacySrgvAdapterTest.java @@ -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 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 writes = - new ArrayList(); + Collections.synchronizedList( + new ArrayList()); + 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 values = new ArrayList(); 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 converted = new ArrayList(); Arrays.stream(values).forEach(converted::add);