This is an automated email from the ASF dual-hosted git repository. chrisdutz pushed a commit to branch develop in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 393044dd51875bf5cc6f1f332fab62be21e1c140 Author: Christofer Dutz <[email protected]> AuthorDate: Thu Aug 13 17:53:28 2026 +0200 fix(plc4j): report unparseable tag addresses per tag --- .../plc4x/java/eip/base/EipTcpConnection.java | 94 +++++-- .../java/modbus/ascii/ModbusAsciiConnection.java | 19 +- .../plc4x/java/modbus/rtu/ModbusRtuConnection.java | 19 +- .../plc4x/java/modbus/tcp/ModbusTcpConnection.java | 19 +- .../rtu/ModbusRtuConnectionRequestChainTest.java | 61 +++++ .../apache/plc4x/java/opcua/OpcuaConnection.java | 93 +++++-- .../plc4x/java/opcua/OpcuaPlcDriverTest.java | 64 +++++ .../org/apache/plc4x/java/s7/S7CotpConnection.java | 43 +++- .../java/s7/optimizer/S7BlockReadOptimizer.java | 3 + .../plc4x/java/s7/optimizer/S7Optimizer.java | 19 +- .../plc4x/java/s7/S7ConnectionInvalidTagTest.java | 106 ++++++++ .../plc4x/java/s7/S7ScriptedConnectionHarness.java | 278 +++++++++++++++++++++ .../simulated/connection/SimulatedConnection.java | 10 + .../connection/SimulatedConnectionTest.java | 56 +++++ 14 files changed, 837 insertions(+), 47 deletions(-) diff --git a/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java b/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java index 3af37c32d3..94e160fe92 100644 --- a/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java +++ b/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java @@ -433,6 +433,14 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi List<CompletableFuture<Void>> tagFutures = new ArrayList<>(); CompletableFuture<Void> chain = CompletableFuture.completedFuture(null); for (String tagName : request.getTagNames()) { + // A tag whose address the builder couldn't parse is kept in the request with an + // error code and a null tag - report that code instead of sending a request built + // from it. + PlcResponseCode requestCode = request.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + values.put(tagName, new DefaultPlcResponseItem<>(requestCode, null)); + continue; + } EipTag eipTag = (EipTag) request.getTag(tagName); CompletableFuture<Void> tagFuture = chain.thenComposeAsync(v -> executeThrottled(() -> { try { @@ -484,8 +492,8 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi PathSegment instanceSegment = new LogicalSegment(new InstanceID((byte) 0, (short) 1)); List<CipService> requests = new ArrayList<>(request.getNumberOfTags()); - for (PlcTag field : request.getTags()) { - EipTag eipTag = (EipTag) field; + for (String tagName : sendableTagNames(request)) { + EipTag eipTag = (EipTag) request.getTag(tagName); try { requests.add(new CipReadRequest(toAnsi(eipTag.getTag()), 1)); } catch (BufferException e) { @@ -527,8 +535,8 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi private CompletableFuture<PlcReadResponse> readWithConnectionManager(PlcReadRequest readRequest) { DefaultPlcReadRequest request = (DefaultPlcReadRequest) readRequest; List<CipService> requests = new ArrayList<>(request.getNumberOfTags()); - for (PlcTag field : request.getTags()) { - EipTag eipTag = (EipTag) field; + for (String tagName : sendableTagNames(request)) { + EipTag eipTag = (EipTag) request.getTag(tagName); try { requests.add(new CipReadRequest(toAnsi(eipTag.getTag()), 1)); } catch (BufferException e) { @@ -564,9 +572,50 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi })); } + /** + * The tag names that can actually be put on the wire. A tag whose address the builder + * couldn't parse stays in the request with an error code and a {@code null} tag; it must + * neither be sent nor occupy a slot in the response, which is mapped back to tags by + * position. Request building and response decoding both filter through this method so the + * two stay aligned. + */ + private static List<String> sendableTagNames(PlcTagRequest request) { + List<String> names = new ArrayList<>(request.getNumberOfTags()); + for (String tagName : request.getTagNames()) { + if (request.getTagResponseCode(tagName) == PlcResponseCode.OK) { + names.add(tagName); + } + } + return names; + } + + /** Response items for the tags the builder rejected, keyed by name. */ + private static Map<String, PlcResponseItem<PlcValue>> rejectedReadTags(PlcTagRequest request) { + Map<String, PlcResponseItem<PlcValue>> rejected = new LinkedHashMap<>(); + for (String tagName : request.getTagNames()) { + PlcResponseCode code = request.getTagResponseCode(tagName); + if (code != PlcResponseCode.OK) { + rejected.put(tagName, new DefaultPlcResponseItem<>(code, null)); + } + } + return rejected; + } + + /** Response codes for the tags the builder rejected, keyed by name. */ + private static Map<String, PlcResponseCode> rejectedWriteTags(PlcTagRequest request) { + Map<String, PlcResponseCode> rejected = new LinkedHashMap<>(); + for (String tagName : request.getTagNames()) { + PlcResponseCode code = request.getTagResponseCode(tagName); + if (code != PlcResponseCode.OK) { + rejected.put(tagName, code); + } + } + return rejected; + } + private Map<String, PlcResponseItem<PlcValue>> errorMap(DefaultPlcReadRequest request) { - Map<String, PlcResponseItem<PlcValue>> values = new LinkedHashMap<>(); - for (String tn : request.getTagNames()) { + Map<String, PlcResponseItem<PlcValue>> values = new LinkedHashMap<>(rejectedReadTags(request)); + for (String tn : sendableTagNames(request)) { values.put(tn, new DefaultPlcResponseItem<>(PlcResponseCode.INTERNAL_ERROR, null)); } return values; @@ -595,7 +644,8 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi List<CompletableFuture<Void>> tagFutures = new ArrayList<>(); CompletableFuture<Void> chain = CompletableFuture.completedFuture(null); - for (String fieldName : request.getTagNames()) { + values.putAll(rejectedWriteTags(request)); + for (String fieldName : sendableTagNames(request)) { EipTag field = (EipTag) request.getTag(fieldName); PlcValue value = request.getPlcValue(fieldName); int elements = Math.max(field.getElementNb(), 1); @@ -645,7 +695,7 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi private CompletableFuture<PlcWriteResponse> writeWithoutConnectionManager(PlcWriteRequest writeRequest) { DefaultPlcWriteRequest request = (DefaultPlcWriteRequest) writeRequest; List<CipWriteRequest> items = new ArrayList<>(writeRequest.getNumberOfTags()); - for (String fieldName : request.getTagNames()) { + for (String fieldName : sendableTagNames(request)) { EipTag field = (EipTag) request.getTag(fieldName); PlcValue value = request.getPlcValue(fieldName); int elements = Math.max(field.getElementNb(), 1); @@ -694,7 +744,7 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi private CompletableFuture<PlcWriteResponse> writeWithConnectionManager(PlcWriteRequest writeRequest) { DefaultPlcWriteRequest request = (DefaultPlcWriteRequest) writeRequest; List<CipWriteRequest> items = new ArrayList<>(writeRequest.getNumberOfTags()); - for (String fieldName : request.getTagNames()) { + for (String fieldName : sendableTagNames(request)) { EipTag field = (EipTag) request.getTag(fieldName); PlcValue value = request.getPlcValue(fieldName); int elements = Math.max(field.getElementNb(), 1); @@ -736,8 +786,8 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi } private Map<String, PlcResponseCode> writeErrorMap(DefaultPlcWriteRequest request) { - Map<String, PlcResponseCode> values = new LinkedHashMap<>(); - for (String tn : request.getTagNames()) { + Map<String, PlcResponseCode> values = new LinkedHashMap<>(rejectedWriteTags(request)); + for (String tn : sendableTagNames(request)) { values.put(tn, PlcResponseCode.INTERNAL_ERROR); } return values; @@ -773,9 +823,13 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi } private PlcReadResponse decodeReadResponse(CipService p, PlcReadRequest readRequest) { - Map<String, PlcResponseItem<PlcValue>> values = new LinkedHashMap<>(); + Map<String, PlcResponseItem<PlcValue>> values = new LinkedHashMap<>(rejectedReadTags(readRequest)); + List<String> sendable = sendableTagNames(readRequest); if (p instanceof CipReadResponse resp) { - String tagName = readRequest.getTagNames().getFirst(); + if (sendable.isEmpty()) { + return new DefaultPlcReadResponse((DefaultPlcReadRequest) readRequest, values); + } + String tagName = sendable.getFirst(); EipTag tag = (EipTag) readRequest.getTag(tagName); PlcResponseCode code = decodeResponseCode(resp.getStatus()); PlcValue plcValue = null; @@ -803,7 +857,7 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi } catch (BufferException e) { throw new PlcRuntimeException(e); } - Iterator<String> it = readRequest.getTagNames().iterator(); + Iterator<String> it = sendable.iterator(); for (int i = 0; i < nb && it.hasNext(); i++) { String tagName = it.next(); EipTag tag = (EipTag) readRequest.getTag(tagName); @@ -916,9 +970,13 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi } private PlcWriteResponse decodeWriteResponse(CipService p, PlcWriteRequest writeRequest) { - Map<String, PlcResponseCode> responses = new LinkedHashMap<>(); + Map<String, PlcResponseCode> responses = new LinkedHashMap<>(rejectedWriteTags(writeRequest)); + List<String> sendable = sendableTagNames(writeRequest); if (p instanceof CipWriteResponse resp) { - String fieldName = writeRequest.getTagNames().getFirst(); + if (sendable.isEmpty()) { + return new DefaultPlcWriteResponse((DefaultPlcWriteRequest) writeRequest, responses); + } + String fieldName = sendable.getFirst(); responses.put(fieldName, decodeResponseCode(resp.getStatus())); return new DefaultPlcWriteResponse(writeRequest, responses); } else if (p instanceof MultipleServiceResponse resp) { @@ -941,7 +999,7 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi } catch (BufferException e) { throw new PlcRuntimeException(e); } - Iterator<String> it = writeRequest.getTagNames().iterator(); + Iterator<String> it = sendable.iterator(); for (int i = 0; i < nb && it.hasNext(); i++) { String fieldName = it.next(); if (arr.get(i) instanceof CipWriteResponse writeResponse) { @@ -950,7 +1008,7 @@ public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfi } return new DefaultPlcWriteResponse(writeRequest, responses); } - for (String tn : writeRequest.getTagNames()) { + for (String tn : sendable) { responses.put(tn, PlcResponseCode.INTERNAL_ERROR); } return new DefaultPlcWriteResponse(writeRequest, responses); diff --git a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java index 41d200abd5..3abf3c3d4e 100644 --- a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java +++ b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java @@ -312,9 +312,17 @@ public class ModbusAsciiConnection extends PollingSubscriptionConnectionBase<Mod protected CompletableFuture<PlcReadResponse> onRead(PlcReadRequest readRequest) { DefaultPlcReadRequest request = (DefaultPlcReadRequest) readRequest; - // Collect all tags + // Collect all tags. A tag whose address the builder couldn't parse is kept in the + // request with an error code and a null tag - it never goes on the wire, its code is + // reported as-is. LinkedHashMap<String, ModbusTag> tagsByName = new LinkedHashMap<>(); + Map<String, PlcResponseItem<PlcValue>> rejectedTags = new LinkedHashMap<>(); for (String tagName : request.getTagNames()) { + PlcResponseCode requestCode = request.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + rejectedTags.put(tagName, new DefaultPlcResponseItem<>(requestCode, null)); + continue; + } tagsByName.put(tagName, (ModbusTag) request.getTag(tagName)); } @@ -341,7 +349,7 @@ public class ModbusAsciiConnection extends PollingSubscriptionConnectionBase<Mod blockFutures.toArray(new CompletableFuture[0])); return allDone.thenApply(v -> { - Map<String, PlcResponseItem<PlcValue>> responseItems = new LinkedHashMap<>(); + Map<String, PlcResponseItem<PlcValue>> responseItems = new LinkedHashMap<>(rejectedTags); for (CompletableFuture<Map<String, PlcResponseItem<PlcValue>>> blockFuture : blockFutures) { try { responseItems.putAll(blockFuture.join()); @@ -400,6 +408,13 @@ public class ModbusAsciiConnection extends PollingSubscriptionConnectionBase<Mod if (request.getTagNames().size() == 1) { String tagName = request.getTagNames().iterator().next(); + // An unparseable address (or value) is kept in the request with an error code and a + // null tag - report that code instead of trying to build a PDU from nothing. + PlcResponseCode requestCode = request.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + return CompletableFuture.completedFuture((PlcWriteResponse) new DefaultPlcWriteResponse( + request, Collections.singletonMap(tagName, requestCode))); + } PlcTag tag = request.getTag(tagName); ModbusPDU requestPdu = getWriteRequestPdu(tag, request.getPlcValue(tagName)); short unitId = getUnitId(tag); diff --git a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java index 9f4bf9df13..04772a3932 100644 --- a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java +++ b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java @@ -312,9 +312,17 @@ public class ModbusRtuConnection extends PollingSubscriptionConnectionBase<Modbu protected CompletableFuture<PlcReadResponse> onRead(PlcReadRequest readRequest) { DefaultPlcReadRequest request = (DefaultPlcReadRequest) readRequest; - // Collect all tags + // Collect all tags. A tag whose address the builder couldn't parse is kept in the + // request with an error code and a null tag - it never goes on the wire, its code is + // reported as-is. LinkedHashMap<String, ModbusTag> tagsByName = new LinkedHashMap<>(); + Map<String, PlcResponseItem<PlcValue>> rejectedTags = new LinkedHashMap<>(); for (String tagName : request.getTagNames()) { + PlcResponseCode requestCode = request.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + rejectedTags.put(tagName, new DefaultPlcResponseItem<>(requestCode, null)); + continue; + } tagsByName.put(tagName, (ModbusTag) request.getTag(tagName)); } @@ -341,7 +349,7 @@ public class ModbusRtuConnection extends PollingSubscriptionConnectionBase<Modbu blockFutures.toArray(new CompletableFuture[0])); return allDone.thenApply(v -> { - Map<String, PlcResponseItem<PlcValue>> responseItems = new LinkedHashMap<>(); + Map<String, PlcResponseItem<PlcValue>> responseItems = new LinkedHashMap<>(rejectedTags); for (CompletableFuture<Map<String, PlcResponseItem<PlcValue>>> blockFuture : blockFutures) { try { responseItems.putAll(blockFuture.join()); @@ -400,6 +408,13 @@ public class ModbusRtuConnection extends PollingSubscriptionConnectionBase<Modbu if (request.getTagNames().size() == 1) { String tagName = request.getTagNames().iterator().next(); + // An unparseable address (or value) is kept in the request with an error code and a + // null tag - report that code instead of trying to build a PDU from nothing. + PlcResponseCode requestCode = request.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + return CompletableFuture.completedFuture((PlcWriteResponse) new DefaultPlcWriteResponse( + request, Collections.singletonMap(tagName, requestCode))); + } PlcTag tag = request.getTag(tagName); ModbusPDU requestPdu = getWriteRequestPdu(tag, request.getPlcValue(tagName)); short unitId = getUnitId(tag); diff --git a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java index 01171b93ad..932180e8c5 100644 --- a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java +++ b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java @@ -217,9 +217,17 @@ public class ModbusTcpConnection extends PollingSubscriptionConnectionBase<Modbu protected CompletableFuture<PlcReadResponse> onRead(PlcReadRequest readRequest) { DefaultPlcReadRequest request = (DefaultPlcReadRequest) readRequest; - // Collect all tags + // Collect all tags. A tag whose address the builder couldn't parse is kept in the + // request with an error code and a null tag - it never goes on the wire, its code is + // reported as-is. LinkedHashMap<String, ModbusTag> tagsByName = new LinkedHashMap<>(); + Map<String, PlcResponseItem<PlcValue>> rejectedTags = new LinkedHashMap<>(); for (String tagName : request.getTagNames()) { + PlcResponseCode requestCode = request.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + rejectedTags.put(tagName, new DefaultPlcResponseItem<>(requestCode, null)); + continue; + } tagsByName.put(tagName, (ModbusTag) request.getTag(tagName)); } @@ -247,7 +255,7 @@ public class ModbusTcpConnection extends PollingSubscriptionConnectionBase<Modbu blockFutures.toArray(new CompletableFuture[0])); return allDone.thenApply(v -> { - Map<String, PlcResponseItem<PlcValue>> responseItems = new LinkedHashMap<>(); + Map<String, PlcResponseItem<PlcValue>> responseItems = new LinkedHashMap<>(rejectedTags); for (CompletableFuture<Map<String, PlcResponseItem<PlcValue>>> blockFuture : blockFutures) { try { responseItems.putAll(blockFuture.join()); @@ -313,6 +321,13 @@ public class ModbusTcpConnection extends PollingSubscriptionConnectionBase<Modbu LinkedHashMap<String, CompletableFuture<PlcResponseCode>> tagFutures = new LinkedHashMap<>(); CompletableFuture<Void> chain = CompletableFuture.completedFuture(null); for (String tagName : request.getTagNames()) { + // An unparseable address (or value) is kept in the request with an error code and a + // null tag - report that code instead of trying to build a PDU from nothing. + PlcResponseCode requestCode = request.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + tagFutures.put(tagName, CompletableFuture.completedFuture(requestCode)); + continue; + } PlcTag tag = request.getTag(tagName); PlcValue value = request.getPlcValue(tagName); CompletableFuture<PlcResponseCode> tagFuture = diff --git a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java index 3767db619e..6c3a1ed395 100644 --- a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java +++ b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java @@ -18,6 +18,9 @@ */ package org.apache.plc4x.java.modbus.rtu; +import org.apache.plc4x.java.api.messages.PlcReadResponse; +import org.apache.plc4x.java.api.messages.PlcWriteResponse; +import org.apache.plc4x.java.api.types.PlcResponseCode; import org.apache.plc4x.java.modbus.readwrite.ModbusErrorCode; import org.apache.plc4x.java.modbus.readwrite.ModbusPDU; import org.apache.plc4x.java.modbus.readwrite.ModbusPDUError; @@ -253,6 +256,64 @@ class ModbusRtuConnectionRequestChainTest { "read must be addressed to the tag's unit id, not the connection default"); } + /** + * An address the tag handler rejects is kept in the request with an error code and a null + * tag. It must be reported as INVALID_ADDRESS and must never reach the wire - reads used to + * silently degrade it to INTERNAL_ERROR, writes threw a NullPointerException while building + * the PDU. + */ + @Test + void readWithInvalidTagAddressIsReportedAndNeverSent() throws Exception { + ScriptedAsyncTransport transport = new ScriptedAsyncTransport(); + ModbusRtuConnection connection = newConnectedConnection(transport); + + PlcReadResponse response = connection.readRequestBuilder() + .addTagAddress("bad", "4x00001:BOGUS") + .build() + .execute() + .get(2, TimeUnit.SECONDS); + + assertEquals(PlcResponseCode.INVALID_ADDRESS, response.getResponseCode("bad")); + assertEquals(0, transport.writeCount(), "a rejected tag must not produce a request"); + } + + @Test + void readMixesValidAndInvalidTagAddresses() throws Exception { + ScriptedAsyncTransport transport = new ScriptedAsyncTransport(); + ModbusRtuConnection connection = newConnectedConnection(transport); + + CompletableFuture<? extends PlcReadResponse> future = connection.readRequestBuilder() + .addTagAddress("good", "4x00001:INT") + .addTagAddress("bad", "4x00001:BOGUS") + .build() + .execute(); + + // The valid tag still goes out on its own and gets answered. + awaitTrue(() -> transport.writeCount() == 1, 2, TimeUnit.SECONDS); + transport.deliver(readResponseFrame(1, new byte[]{0x00, 0x2A})); + transport.runDataListener(); + + PlcReadResponse response = future.get(2, TimeUnit.SECONDS); + assertEquals(PlcResponseCode.OK, response.getResponseCode("good")); + assertEquals(42, response.getInteger("good")); + assertEquals(PlcResponseCode.INVALID_ADDRESS, response.getResponseCode("bad")); + } + + @Test + void writeWithInvalidTagAddressIsReportedAndNeverSent() throws Exception { + ScriptedAsyncTransport transport = new ScriptedAsyncTransport(); + ModbusRtuConnection connection = newConnectedConnection(transport); + + PlcWriteResponse response = connection.writeRequestBuilder() + .addTagAddress("bad", "4x00001:BOGUS", 42) + .build() + .execute() + .get(2, TimeUnit.SECONDS); + + assertEquals(PlcResponseCode.INVALID_ADDRESS, response.getResponseCode("bad")); + assertEquals(0, transport.writeCount(), "a rejected tag must not produce a request"); + } + /** * Builds a connection wired to the given fake transport, and drives it * through the same construction/connect path as {@code ModbusRtuConnectionTest}. diff --git a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java index 0d132563cc..973b2632a4 100644 --- a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java +++ b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java @@ -321,12 +321,17 @@ public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration> implemen DefaultPlcReadRequest request = (DefaultPlcReadRequest) readRequest; RequestHeader requestHeader = conversation.createRequestHeader(); - List<ReadValueId> readValueArray = new ArrayList<>(request.getTagNames().size()); - Iterator<String> iterator = request.getTagNames().iterator(); + // Tags the builder rejected (unparseable address) carry their code straight into the + // response; they have no node id to ask the server for. + Map<String, PlcResponseItem<PlcValue>> rejectedTags = rejectedReadTags(request); + List<String> sendable = sendableTagNames(request); + if (sendable.isEmpty()) { + return CompletableFuture.completedFuture(new DefaultPlcReadResponse(request, rejectedTags)); + } + + List<ReadValueId> readValueArray = new ArrayList<>(sendable.size()); Map<String, PlcTag> tagMap = new LinkedHashMap<>(); - for (int i = 0; i < request.getTagNames().size(); i++) { - String tagName = iterator.next(); - // TODO: We need to check that the tag-return-code is OK as it could also be INVALID_TAG + for (String tagName : sendable) { OpcuaTag tag = (OpcuaTag) request.getTag(tagName); tagMap.put(tagName, tag); @@ -362,17 +367,58 @@ public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration> implemen } if (structFutures.isEmpty()) { return CompletableFuture.completedFuture(new DefaultPlcReadResponse(request, - readResponse(tagMap, results, Collections.emptyMap()))); + inRequestOrder(request, readResponse(tagMap, results, Collections.emptyMap()), rejectedTags))); } return CompletableFuture.allOf(structFutures.values().toArray(new CompletableFuture[0])) .thenApply(v -> { Map<String, StructureDefinition> structDefs = new HashMap<>(); structFutures.forEach((name, future) -> structDefs.put(name, future.getNow(null))); - return new DefaultPlcReadResponse(request, readResponse(tagMap, results, structDefs)); + return new DefaultPlcReadResponse(request, + inRequestOrder(request, readResponse(tagMap, results, structDefs), rejectedTags)); }); }); } + /** + * The tag names that can actually be put on the wire. A tag whose address the builder + * couldn't parse stays in the request with an error code and a {@code null} tag; it must + * neither be sent nor take up a slot in the response, which OPC UA maps back to tags by + * position. + */ + private static List<String> sendableTagNames(PlcTagRequest request) { + List<String> names = new ArrayList<>(request.getNumberOfTags()); + for (String tagName : request.getTagNames()) { + if (request.getTagResponseCode(tagName) == PlcResponseCode.OK) { + names.add(tagName); + } + } + return names; + } + + /** Response items for the tags the builder rejected, keyed by name. */ + private static Map<String, PlcResponseItem<PlcValue>> rejectedReadTags(PlcTagRequest request) { + Map<String, PlcResponseItem<PlcValue>> rejected = new LinkedHashMap<>(); + for (String tagName : request.getTagNames()) { + PlcResponseCode code = request.getTagResponseCode(tagName); + if (code != PlcResponseCode.OK) { + rejected.put(tagName, new DefaultPlcResponseItem<>(code, null)); + } + } + return rejected; + } + + /** Response codes for the tags the builder rejected, keyed by name. */ + private static Map<String, PlcResponseCode> rejectedWriteTags(PlcTagRequest request) { + Map<String, PlcResponseCode> rejected = new LinkedHashMap<>(); + for (String tagName : request.getTagNames()) { + PlcResponseCode code = request.getTagResponseCode(tagName); + if (code != PlcResponseCode.OK) { + rejected.put(tagName, code); + } + } + return rejected; + } + /** The tag's OPC UA IndexRange as a PascalString, or the null string when the whole node is addressed. */ private static PascalString indexRangeOf(OpcuaTag tag) { String indexRange = tag.getIndexRange(); @@ -401,6 +447,18 @@ public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration> implemen return nodeId; } + /** Merges decoded and rejected tags back into the order the caller asked for. */ + private static Map<String, PlcResponseItem<PlcValue>> inRequestOrder(PlcTagRequest request, + Map<String, PlcResponseItem<PlcValue>> decoded, + Map<String, PlcResponseItem<PlcValue>> rejected) { + Map<String, PlcResponseItem<PlcValue>> ordered = new LinkedHashMap<>(); + for (String tagName : request.getTagNames()) { + PlcResponseItem<PlcValue> item = decoded.get(tagName); + ordered.put(tagName, item != null ? item : rejected.get(tagName)); + } + return ordered; + } + public Map<String, PlcResponseItem<PlcValue>> readResponse(Map<String, PlcTag> tagMap, List<DataValue> results) { return readResponse(tagMap, results, Collections.emptyMap()); } @@ -832,7 +890,7 @@ public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration> implemen private CompletableFuture<Map<String, StructWriteInfo>> resolveStructWriteInfos(DefaultPlcWriteRequest request) { Map<String, StructWriteInfo> resolved = new ConcurrentHashMap<>(); List<CompletableFuture<?>> futures = new ArrayList<>(); - for (String tagName : request.getTagNames()) { + for (String tagName : sendableTagNames(request)) { if (!isStructValue(request.getPlcValue(tagName))) { continue; } @@ -2203,14 +2261,19 @@ public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration> implemen LOGGER.trace("Writing Value"); DefaultPlcWriteRequest request = (DefaultPlcWriteRequest) writeRequest; + if (sendableTagNames(request).isEmpty()) { + return CompletableFuture.completedFuture( + new DefaultPlcWriteResponse(request, rejectedWriteTags(request))); + } + // Phase 4: for tags without an explicit ;TYPE suffix, resolve the server-declared data // type (via the session type cache) up-front so the write is built with the authoritative // OPC UA type instead of a lossy Java-value guess. Then assemble and submit the write. return resolveWriteTypes(request).thenCompose(serverAttributes -> resolveStructWriteInfos(request).thenCompose(structInfos -> { RequestHeader requestHeader = conversation.createRequestHeader(); - List<WriteValue> writeValueList = new ArrayList<>(request.getTagNames().size()); - for (String tagName : request.getTagNames()) { + List<WriteValue> writeValueList = new ArrayList<>(request.getNumberOfTags()); + for (String tagName : sendableTagNames(request)) { OpcuaTag tag = (OpcuaTag) request.getTag(tagName); NodeId nodeId = generateNodeId(tag); @@ -2251,7 +2314,7 @@ public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration> implemen private CompletableFuture<Map<String, NodeAttributes>> resolveWriteTypes(DefaultPlcWriteRequest request) { Map<String, NodeAttributes> serverAttributes = new ConcurrentHashMap<>(); List<CompletableFuture<?>> futures = new ArrayList<>(); - for (String tagName : request.getTagNames()) { + for (String tagName : sendableTagNames(request)) { OpcuaTag tag = (OpcuaTag) request.getTag(tagName); // An explicit ;TYPE suffix is authoritative — no server round-trip needed. if (tag.getDataType() != OpcuaDataType.NULL) { @@ -2274,10 +2337,12 @@ public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration> implemen } private PlcWriteResponse writeResponse(DefaultPlcWriteRequest request, WriteResponse writeResponse) { - Map<String, PlcResponseCode> responseMap = new HashMap<>(); + Map<String, PlcResponseCode> responseMap = new HashMap<>(rejectedWriteTags(request)); List<StatusCode> results = writeResponse.getResults(); - Iterator<String> responseIterator = request.getTagNames().iterator(); - for (int i = 0; i < request.getTagNames().size(); i++) { + // Only the tags that were actually sent have a slot in the response. + List<String> sendable = sendableTagNames(request); + Iterator<String> responseIterator = sendable.iterator(); + for (int i = 0; i < sendable.size(); i++) { String tagName = responseIterator.next(); long opcStatusCode = results.get(i).getStatusCode(); PlcResponseCode statusCode = mapOpcStatusCode(opcStatusCode, PlcResponseCode.REMOTE_ERROR); diff --git a/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java b/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java index b100023b40..dc2fcdea7c 100644 --- a/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java +++ b/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java @@ -215,6 +215,70 @@ public class OpcuaPlcDriverTest { connectionStringValidSet = List.of(tcpConnectionAddress); } + /** + * A tag whose address the tag handler can't parse stays in the request carrying an error code + * and a {@code null} tag. It has to come back as INVALID_ADDRESS - reading it used to throw a + * NullPointerException while the driver built the node id (there was a TODO in the read path + * admitting the check was missing). + */ + @Test + void readWithInvalidTagAddressIsReported() throws Exception { + try (PlcConnection connection = new DefaultPlcDriverManager().getConnection(tcpConnectionAddress)) { + PlcReadResponse response = connection.readRequestBuilder() + .addTagAddress("bad", "this-is-not-a-node-id") + .build().execute().get(30, TimeUnit.SECONDS); + + assertThat(response.getResponseCode("bad")).isEqualTo(PlcResponseCode.INVALID_ADDRESS); + } + } + + /** + * The point of per-tag codes: one bad address must not cost the caller the other tags of the + * same request, and the good values must not be shifted onto the wrong names - OPC UA maps + * its results back to tags by position. + */ + @Test + void readMixesValidAndInvalidTagAddresses() throws Exception { + try (PlcConnection connection = new DefaultPlcDriverManager().getConnection(tcpConnectionAddress)) { + PlcReadResponse response = connection.readRequestBuilder() + .addTagAddress("bad", "this-is-not-a-node-id") + .addTagAddress("bool", BOOL_IDENTIFIER_READ_WRITE) + .addTagAddress("int32", INT32_IDENTIFIER_READ_WRITE) + .build().execute().get(30, TimeUnit.SECONDS); + + assertThat(response.getResponseCode("bad")).isEqualTo(PlcResponseCode.INVALID_ADDRESS); + assertThat(response.getResponseCode("bool")).isEqualTo(PlcResponseCode.OK); + assertThat(response.getResponseCode("int32")).isEqualTo(PlcResponseCode.OK); + // Values must still belong to the tag that asked for them. + assertThat(response.getPlcValue("bool").isBoolean()).isTrue(); + assertThat(response.getPlcValue("int32").isLong() || response.getPlcValue("int32").isInteger()).isTrue(); + } + } + + @Test + void writeWithInvalidTagAddressIsReported() throws Exception { + try (PlcConnection connection = new DefaultPlcDriverManager().getConnection(tcpConnectionAddress)) { + PlcWriteResponse response = connection.writeRequestBuilder() + .addTagAddress("bad", "this-is-not-a-node-id", 42) + .build().execute().get(30, TimeUnit.SECONDS); + + assertThat(response.getResponseCode("bad")).isEqualTo(PlcResponseCode.INVALID_ADDRESS); + } + } + + @Test + void writeMixesValidAndInvalidTagAddresses() throws Exception { + try (PlcConnection connection = new DefaultPlcDriverManager().getConnection(tcpConnectionAddress)) { + PlcWriteResponse response = connection.writeRequestBuilder() + .addTagAddress("bad", "this-is-not-a-node-id", 42) + .addTagAddress("bool", BOOL_IDENTIFIER_READ_WRITE, true) + .build().execute().get(30, TimeUnit.SECONDS); + + assertThat(response.getResponseCode("bad")).isEqualTo(PlcResponseCode.INVALID_ADDRESS); + assertThat(response.getResponseCode("bool")).isEqualTo(PlcResponseCode.OK); + } + } + @Test void browseWildcardDiscoversWholeAddressSpace() throws Exception { try (PlcConnection connection = new DefaultPlcDriverManager().getConnection(tcpConnectionAddress)) { diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/S7CotpConnection.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/S7CotpConnection.java index 4b30f3c028..fafe798fa7 100644 --- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/S7CotpConnection.java +++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/S7CotpConnection.java @@ -418,14 +418,26 @@ public class S7CotpConnection extends ConnectionBase<S7Configuration> { @Override protected CompletableFuture<PlcReadResponse> onRead(PlcReadRequest readRequest) { - // Validate tag types up front so we fail fast. + // A tag whose address the builder couldn't parse stays in the request with an error code + // and a null tag. It is reported per tag rather than failing the whole request, so one + // typo doesn't stop the request's other tags from being read. + Map<String, PlcResponseItem<PlcValue>> rejectedTags = new LinkedHashMap<>(); for (String tagName : readRequest.getTagNames()) { + PlcResponseCode requestCode = readRequest.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + rejectedTags.put(tagName, new DefaultPlcResponseItem<>(requestCode, null)); + continue; + } org.apache.plc4x.java.api.model.PlcTag tag = readRequest.getTag(tagName); if (!(tag instanceof S7Tag)) { + // Not a parse failure but a tag object of the wrong type handed to the API. return CompletableFuture.failedFuture( - new PlcProtocolException("Unsupported tag type " + (tag == null ? "null" : tag.getClass().getName()))); + new PlcProtocolException("Unsupported tag type " + tag.getClass().getName())); } } + if (rejectedTags.size() == readRequest.getTagNames().size()) { + return CompletableFuture.completedFuture(new DefaultPlcReadResponse(readRequest, rejectedTags)); + } List<S7ReadChunk> chunks; try { @@ -434,7 +446,8 @@ public class S7CotpConnection extends ConnectionBase<S7Configuration> { // A tag the optimizer can't fit at all is reported per-tag rather than aborting the whole request. Map<String, PlcResponseItem<PlcValue>> values = new LinkedHashMap<>(); for (String t : readRequest.getTagNames()) { - values.put(t, new DefaultPlcResponseItem<>(PlcResponseCode.INVALID_DATA, null)); + values.put(t, rejectedTags.getOrDefault(t, + new DefaultPlcResponseItem<>(PlcResponseCode.INVALID_DATA, null))); } return CompletableFuture.completedFuture(new DefaultPlcReadResponse(readRequest, values)); } @@ -445,9 +458,10 @@ public class S7CotpConnection extends ConnectionBase<S7Configuration> { // For each chunk, send one S7 read message; combine the per-chunk decoders into the final response. Map<String, PlcResponseItem<PlcValue>> finalValues = new LinkedHashMap<>(); - // Pre-seed in original request order so output order is deterministic. + // Pre-seed in original request order so output order is deterministic. Tags the builder + // rejected already have their code and are never overwritten (only nulls are). for (String t : readRequest.getTagNames()) { - finalValues.put(t, null); + finalValues.put(t, rejectedTags.get(t)); } CompletableFuture<Void> chain = CompletableFuture.completedFuture(null); for (S7ReadChunk chunk : chunks) { @@ -597,20 +611,33 @@ public class S7CotpConnection extends ConnectionBase<S7Configuration> { @Override protected CompletableFuture<PlcWriteResponse> onWrite(PlcWriteRequest writeRequest) { + // Same as for reads: a tag the builder rejected is reported per tag, not by failing the + // whole request. + Map<String, PlcResponseCode> rejectedTags = new LinkedHashMap<>(); for (String tagName : writeRequest.getTagNames()) { + PlcResponseCode requestCode = writeRequest.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + rejectedTags.put(tagName, requestCode); + continue; + } org.apache.plc4x.java.api.model.PlcTag tag = writeRequest.getTag(tagName); if (!(tag instanceof S7Tag)) { return CompletableFuture.failedFuture(new PlcProtocolException( - "Unsupported tag type " + (tag == null ? "null" : tag.getClass().getName()))); + "Unsupported tag type " + tag.getClass().getName())); } } + if (rejectedTags.size() == writeRequest.getTagNames().size()) { + return CompletableFuture.completedFuture(new DefaultPlcWriteResponse(writeRequest, rejectedTags)); + } List<S7WriteChunk> chunks; try { chunks = optimizer.splitWriteRequest(writeRequest, driverContext); } catch (PlcRuntimeException e) { Map<String, PlcResponseCode> codes = new LinkedHashMap<>(); - for (String t : writeRequest.getTagNames()) codes.put(t, PlcResponseCode.INVALID_DATA); + for (String t : writeRequest.getTagNames()) { + codes.put(t, rejectedTags.getOrDefault(t, PlcResponseCode.INVALID_DATA)); + } return CompletableFuture.completedFuture(new DefaultPlcWriteResponse(writeRequest, codes)); } if (chunks.isEmpty()) { @@ -619,7 +646,7 @@ public class S7CotpConnection extends ConnectionBase<S7Configuration> { Map<String, PlcResponseCode> finalCodes = new LinkedHashMap<>(); for (String t : writeRequest.getTagNames()) { - finalCodes.put(t, null); + finalCodes.put(t, rejectedTags.get(t)); } CompletableFuture<Void> chain = CompletableFuture.completedFuture(null); for (S7WriteChunk chunk : chunks) { diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7BlockReadOptimizer.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7BlockReadOptimizer.java index 0ea80c8c5d..e2ffc36bdb 100644 --- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7BlockReadOptimizer.java +++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7BlockReadOptimizer.java @@ -50,6 +50,9 @@ public class S7BlockReadOptimizer extends S7Optimizer { Map<String, List<TagEntry>> tagsPerArea = new LinkedHashMap<>(); LinkedHashMap<String, PlcTag> passthrough = new LinkedHashMap<>(); for (String tagName : request.getTagNames()) { + if (isRejected(request, tagName)) { + continue; + } PlcTag plcTag = request.getTag(tagName); if (!(plcTag instanceof S7Tag s7Tag) || plcTag instanceof S7StringVarLengthTag) { // Block-merging is unsafe for var-length strings (response size is dynamic); diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7Optimizer.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7Optimizer.java index e5a0a7b4eb..333d0cc658 100644 --- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7Optimizer.java +++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/optimizer/S7Optimizer.java @@ -20,6 +20,8 @@ package org.apache.plc4x.java.s7.optimizer; import org.apache.plc4x.java.api.exceptions.PlcRuntimeException; import org.apache.plc4x.java.api.messages.PlcReadRequest; +import org.apache.plc4x.java.api.messages.PlcTagRequest; +import org.apache.plc4x.java.api.types.PlcResponseCode; import org.apache.plc4x.java.api.messages.PlcWriteRequest; import org.apache.plc4x.java.api.model.PlcTag; import org.apache.plc4x.java.s7.readwrite.MemoryArea; @@ -32,7 +34,6 @@ import org.apache.plc4x.java.s7.readwrite.S7ParameterWriteVarRequest; import org.apache.plc4x.java.s7.readwrite.S7ParameterWriteVarResponse; import org.apache.plc4x.java.s7.readwrite.S7PayloadReadVarResponse; import org.apache.plc4x.java.s7.readwrite.S7PayloadWriteVarResponse; -import org.apache.plc4x.java.s7.readwrite.S7VarRequestParameterItem; import org.apache.plc4x.java.s7.readwrite.S7VarRequestParameterItemAddress; import org.apache.plc4x.java.s7.readwrite.TransportSize; import org.apache.plc4x.java.s7.context.S7DriverContext; @@ -157,6 +158,9 @@ public class S7Optimizer { int curResponseSize = EMPTY_WRITE_RESPONSE_SIZE; for (String tagName : request.getTagNames()) { + if (isRejected(request, tagName)) { + continue; + } PlcTag plcTag = request.getTag(tagName); if (!(plcTag instanceof S7Tag s7Tag)) { throw new PlcRuntimeException("Unsupported tag type for tag " + tagName); @@ -254,8 +258,21 @@ public class S7Optimizer { protected static LinkedHashMap<String, PlcTag> toLinkedMap(PlcReadRequest request) { LinkedHashMap<String, PlcTag> out = new LinkedHashMap<>(); for (String n : request.getTagNames()) { + if (isRejected(request, n)) { + // Address the builder couldn't parse: it has no tag to chunk, and the + // connection reports its code directly. + continue; + } out.put(n, request.getTag(n)); } return out; } + + /** + * Whether the request builder already rejected this tag (unparseable address, bad value), + * in which case it carries an error code and a {@code null} tag. + */ + protected static boolean isRejected(PlcTagRequest request, String tagName) { + return request.getTagResponseCode(tagName) != PlcResponseCode.OK; + } } diff --git a/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/S7ConnectionInvalidTagTest.java b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/S7ConnectionInvalidTagTest.java new file mode 100644 index 0000000000..57c56b64ad --- /dev/null +++ b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/S7ConnectionInvalidTagTest.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.plc4x.java.s7; + +import org.apache.plc4x.java.api.messages.PlcReadResponse; +import org.apache.plc4x.java.api.messages.PlcWriteResponse; +import org.apache.plc4x.java.api.types.PlcResponseCode; +import org.apache.plc4x.java.s7.S7ScriptedConnectionHarness.ScriptedS7Transport; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A tag whose address the builder couldn't parse stays in the request with an error code and a + * null tag. It has to be reported per tag as INVALID_ADDRESS - the driver used to fail the + * whole request with a PlcProtocolException, which took the request's valid tags down with it. + */ +class S7ConnectionInvalidTagTest { + + @Test + void harnessBringsUpAConnection() throws Exception { + ScriptedS7Transport transport = new ScriptedS7Transport(); + S7CotpConnection connection = S7ScriptedConnectionHarness.newConnectedConnection(transport); + + assertTrue(connection.isConnected(), "the scripted handshake should leave the connection up"); + connection.close(); + } + + @Test + void readWithInvalidTagAddressIsReportedAndNeverSent() throws Exception { + ScriptedS7Transport transport = new ScriptedS7Transport(); + S7CotpConnection connection = S7ScriptedConnectionHarness.newConnectedConnection(transport); + + PlcReadResponse response = connection.readRequestBuilder() + .addTagAddress("bad", "%DB1:NOSUCHTHING") + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + assertEquals(PlcResponseCode.INVALID_ADDRESS, response.getResponseCode("bad")); + assertEquals(0, transport.writeCount(), "a rejected tag must not produce a request"); + + connection.close(); + } + + @Test + void writeWithInvalidTagAddressIsReportedAndNeverSent() throws Exception { + ScriptedS7Transport transport = new ScriptedS7Transport(); + S7CotpConnection connection = S7ScriptedConnectionHarness.newConnectedConnection(transport); + + PlcWriteResponse response = connection.writeRequestBuilder() + .addTagAddress("bad", "%DB1:NOSUCHTHING", 42) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + assertEquals(PlcResponseCode.INVALID_ADDRESS, response.getResponseCode("bad")); + assertEquals(0, transport.writeCount(), "a rejected tag must not produce a request"); + + connection.close(); + } + + /** + * The point of per-tag codes: one typo must not stop the rest of the request from being + * read. The valid tag has to reach the wire even though a sibling was rejected. + */ + @Test + void readMixesValidAndInvalidTagAddresses() throws Exception { + ScriptedS7Transport transport = new ScriptedS7Transport(); + S7CotpConnection connection = S7ScriptedConnectionHarness.newConnectedConnection(transport); + + connection.readRequestBuilder() + .addTagAddress("good", "%DB1.DBW0:INT") + .addTagAddress("bad", "%DB1:NOSUCHTHING") + .build() + .execute(); + + // The valid tag still goes out on its own. + long deadline = System.currentTimeMillis() + 5_000; + while (transport.writeCount() == 0 && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertEquals(1, transport.writeCount(), "the valid tag must still be requested"); + + connection.close(); + } +} diff --git a/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/S7ScriptedConnectionHarness.java b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/S7ScriptedConnectionHarness.java new file mode 100644 index 0000000000..a8408074ba --- /dev/null +++ b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/S7ScriptedConnectionHarness.java @@ -0,0 +1,278 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.plc4x.java.s7; + +import org.apache.plc4x.java.s7.configuration.S7Configuration; +import org.apache.plc4x.java.s7.readwrite.ControllerType; +import org.apache.plc4x.java.s7.readwrite.S7Message; +import org.apache.plc4x.java.s7.readwrite.S7MessageResponseData; +import org.apache.plc4x.java.s7.readwrite.S7ParameterSetupCommunication; +import org.apache.plc4x.java.spi.buffers.api.WithOption; +import org.apache.plc4x.java.spi.buffers.bytebased.ReadBufferByteBased; +import org.apache.plc4x.java.spi.buffers.bytebased.WithByteBasedOption; +import org.apache.plc4x.java.spi.buffers.bytebased.WriteBufferByteBased; +import org.apache.plc4x.java.spi.transports.api.AsyncTransportInstance; +import org.apache.plc4x.java.spi.transports.api.config.TransportConfiguration; +import org.apache.plc4x.java.spi.transports.api.exceptions.TransportException; +import org.apache.plc4x.java.utils.auditlog.api.AuditLog; + +import java.io.ByteArrayOutputStream; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Brings up a real {@link S7CotpConnection} against an in-memory transport, so behaviour that + * only shows up on a connected driver (per-tag response codes, what actually reaches the wire) + * can be tested without a PLC. + * <p> + * Connecting an S7 driver is a three step conversation - COTP connection request, S7 setup + * communication, and an SZL identification probe - and the connection blocks until it is done. + * A responder thread answers each outgoing frame with a canned reply, which is enough to get + * the connection into its normal operating state. + */ +public final class S7ScriptedConnectionHarness { + + private S7ScriptedConnectionHarness() { + } + + /** + * Connects an S7 connection wired to the given transport, scripting the handshake. + */ + public static S7CotpConnection newConnectedConnection(ScriptedS7Transport transport) throws Exception { + S7Configuration configuration = new S7Configuration(); + // Pinning the controller type makes the driver skip the SZL capability probe, so the + // handshake is just COTP connect + S7 setup communication. + configuration.setControllerType(ControllerType.S7_300); + S7CotpConnection connection = new S7CotpConnection(configuration, transport, + AuditLog.builder().build()); + + // The handshake blocks, so it has to run while the responder feeds it answers. + AtomicReference<Exception> failure = new AtomicReference<>(); + Thread connectThread = new Thread(() -> { + try { + connection.connect(); + } catch (Exception e) { + failure.set(e); + } + }, "s7-harness-connect"); + connectThread.setDaemon(true); + + AtomicBoolean handshakeDone = new AtomicBoolean(false); + Thread responder = new Thread(() -> { + int answered = 0; + while (!handshakeDone.get()) { + if (transport.writeCount() > answered) { + byte[] reply = replyFor(answered, transport.writtenFrames().get(answered)); + answered++; + if (reply != null) { + if (Boolean.getBoolean("s7harness.debug")) { + System.out.println("HARNESS: replying to frame " + (answered - 1) + + " with " + java.util.HexFormat.of().formatHex(reply) + + " (listener registered: " + transport.hasDataListener() + ")"); + } + transport.deliver(reply); + transport.runDataListener(); + } + } + try { + Thread.sleep(5); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + }, "s7-harness-responder"); + responder.setDaemon(true); + + responder.start(); + connectThread.start(); + connectThread.join(15_000); + handshakeDone.set(true); + + if (failure.get() != null) { + throw failure.get(); + } + if (connectThread.isAlive()) { + throw new IllegalStateException("S7 handshake did not complete"); + } + transport.resetCounters(); + return connection; + } + + /** + * The canned reply for the n-th frame the driver sends during the handshake. The third + * frame is the SZL capability probe, which the driver is designed to survive failing - it + * just marks UserData services unavailable - so it is left unanswered. + */ + private static byte[] replyFor(int frameIndex, byte[] request) { + try { + if (frameIndex == 0) { + // Responses are correlated by tpdu reference, so echo the one the driver used. + return s7SetupCommunicationResponse(tpduReferenceOf(request)); + } + return null; + } catch (Exception e) { + throw new IllegalStateException("Could not build handshake reply " + frameIndex, e); + } + } + + /** Digs the S7 tpdu reference out of a frame the driver just sent. */ + private static int tpduReferenceOf(byte[] frame) throws Exception { + return S7Message.staticParse(readBuffer(frame)).getTpduReference(); + } + + private static byte[] s7SetupCommunicationResponse(int tpduReference) throws Exception { + S7Message message = new S7MessageResponseData(tpduReference, + new S7ParameterSetupCommunication(8, 8, 240), null, (short) 0, (short) 0); + return wireBytes(message); + } + + /** + * The COTP/TPKT framing is done by the transport this driver sits on, so what the driver + * itself reads and writes are plain S7 messages. + */ + private static byte[] wireBytes(S7Message message) throws Exception { + WriteBufferByteBased writeBuffer = new WriteBufferByteBased(new byte[message.getLengthInBytes()], + ENCODING_OPTIONS); + message.serialize(writeBuffer); + return writeBuffer.getBytes(); + } + + private static ReadBufferByteBased readBuffer(byte[] data) { + return new ReadBufferByteBased(data, ENCODING_OPTIONS); + } + + /** + * The generated S7 parsers/serializers require these to be set on the buffer - the driver's + * message codec configures the very same ones. + */ + private static final WithOption[] ENCODING_OPTIONS = { + WithOption.WithUnsignedIntegerEncoding("unsigned-binary"), + WithOption.WithSignedIntegerEncoding("twos-complement"), + WithOption.WithFloatEncoding("IEEE754"), + WithByteBasedOption.WithByteOrder("BIG_ENDIAN") + }; + + /** + * In-memory transport double: records everything the driver writes and lets a test feed + * bytes back in. Mirrors the scripted transport used by the Modbus tests. + */ + public static final class ScriptedS7Transport implements AsyncTransportInstance<TransportConfiguration> { + private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + private int readPosition; + private boolean open = true; + private final AtomicInteger writeCount = new AtomicInteger(); + private final List<byte[]> writtenFrames = new CopyOnWriteArrayList<>(); + private final AtomicReference<Runnable> dataListener = new AtomicReference<>(); + + public void deliver(byte[] bytes) { + synchronized (buffer) { + buffer.writeBytes(bytes); + } + } + + public int writeCount() { + return writeCount.get(); + } + + public List<byte[]> writtenFrames() { + return writtenFrames; + } + + /** Forgets the handshake traffic so a test only sees what its own request produced. */ + public void resetCounters() { + writeCount.set(0); + writtenFrames.clear(); + } + + public boolean hasDataListener() { + return dataListener.get() != null; + } + + public void runDataListener() { + Runnable listener = dataListener.get(); + if (listener != null) { + listener.run(); + } + } + + @Override + public TransportConfiguration getConfiguration() { + return null; + } + + @Override + public boolean isOpen() { + return open; + } + + @Override + public int getNumBytesAvailable() { + synchronized (buffer) { + return buffer.size() - readPosition; + } + } + + @Override + public byte[] peekReadableBytes(int numBytes) throws TransportException { + synchronized (buffer) { + if (numBytes > getNumBytesAvailable()) { + throw new TransportException("peek beyond available: " + numBytes); + } + byte[] all = buffer.toByteArray(); + byte[] result = new byte[numBytes]; + System.arraycopy(all, readPosition, result, 0, numBytes); + return result; + } + } + + @Override + public byte[] read(int numBytes) throws TransportException { + synchronized (buffer) { + byte[] result = peekReadableBytes(numBytes); + readPosition += numBytes; + return result; + } + } + + @Override + public void write(byte[] bytes) { + writtenFrames.add(bytes.clone()); + writeCount.incrementAndGet(); + } + + @Override + public void close() { + open = false; + } + + @Override + public void registerDataListener(Runnable listener) { + dataListener.set(listener); + } + + @Override + public void removeDataListener() { + dataListener.set(null); + } + } +} diff --git a/plc4j/drivers/simulated/src/main/java/org/apache/plc4x/java/simulated/connection/SimulatedConnection.java b/plc4j/drivers/simulated/src/main/java/org/apache/plc4x/java/simulated/connection/SimulatedConnection.java index a457cf36be..9225312df1 100644 --- a/plc4j/drivers/simulated/src/main/java/org/apache/plc4x/java/simulated/connection/SimulatedConnection.java +++ b/plc4j/drivers/simulated/src/main/java/org/apache/plc4x/java/simulated/connection/SimulatedConnection.java @@ -155,6 +155,13 @@ public class SimulatedConnection extends ConnectionBase<SimulatedConfiguration> protected CompletableFuture<PlcReadResponse> onRead(PlcReadRequest readRequest) { Map<String, PlcResponseItem<PlcValue>> tags = new HashMap<>(); for (String tagName : readRequest.getTagNames()) { + // A tag the builder couldn't parse is kept in the request with its error code and a + // null tag, so echo that code instead of dereferencing the tag (as onWrite does). + PlcResponseCode requestCode = readRequest.getTagResponseCode(tagName); + if (requestCode != PlcResponseCode.OK) { + tags.put(tagName, new DefaultPlcResponseItem<>(requestCode, null)); + continue; + } SimulatedTag tag = (SimulatedTag) readRequest.getTag(tagName); Optional<PlcValue> value = device.get(tag); tags.put(tagName, value @@ -189,6 +196,9 @@ public class SimulatedConnection extends ConnectionBase<SimulatedConfiguration> protected CompletableFuture<PlcSubscriptionResponse> onSubscribe(PlcSubscriptionRequest subscriptionRequest) { Map<String, PlcResponseItem<PlcSubscriptionHandle>> values = new LinkedHashMap<>(); for (String name : subscriptionRequest.getTagNames()) { + // No invalid-address guard needed here: unlike the read/write builders, the + // subscription builder rejects an unparseable address by throwing, so a request + // that reaches this point only holds tags that parsed. SimulatedSubscriptionHandle handle = new SimulatedSubscriptionHandle(this, name); DefaultPlcSubscriptionTag subscriptionTag = (DefaultPlcSubscriptionTag) subscriptionRequest.getTag(name); diff --git a/plc4j/drivers/simulated/src/test/java/org/apache/plc4x/java/simulated/connection/SimulatedConnectionTest.java b/plc4j/drivers/simulated/src/test/java/org/apache/plc4x/java/simulated/connection/SimulatedConnectionTest.java index 632a301840..832187ede0 100644 --- a/plc4j/drivers/simulated/src/test/java/org/apache/plc4x/java/simulated/connection/SimulatedConnectionTest.java +++ b/plc4j/drivers/simulated/src/test/java/org/apache/plc4x/java/simulated/connection/SimulatedConnectionTest.java @@ -19,6 +19,10 @@ package org.apache.plc4x.java.simulated.connection; import org.apache.plc4x.java.api.messages.*; +import org.apache.plc4x.java.api.exceptions.PlcInvalidTagException; +import org.apache.plc4x.java.api.types.PlcResponseCode; +import org.apache.plc4x.java.simulated.tag.SimulatedTag; +import org.apache.plc4x.java.spi.values.PlcSTRING; import org.apache.plc4x.java.api.model.PlcConsumerRegistration; import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; import org.assertj.core.api.WithAssertions; @@ -34,13 +38,16 @@ import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.Collection; import java.util.Collections; +import java.util.Optional; import java.util.Queue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) class SimulatedConnectionTest implements WithAssertions { @@ -102,6 +109,41 @@ class SimulatedConnectionTest implements WithAssertions { PlcWriteResponse response = write.get(1, TimeUnit.SECONDS); assertThat(response).isNotNull(); } + + /** + * An address the tag handler can't parse is kept in the request with an error code and a + * null tag. Reading it must report that code, not fail with a NullPointerException. + * ("Boolean" is not a PlcValueType - "BOOL" is.) + */ + @Test + void readWithInvalidTagAddress() throws Exception { + PlcReadRequest plcReadRequest = SUT.readRequestBuilder() + .addTagAddress("invalid", "RANDOM/foo:Boolean") + .build(); + + PlcReadResponse response = SUT.read(plcReadRequest).get(1, TimeUnit.SECONDS); + + assertThat(response).isNotNull(); + assertThat(response.getResponseCode("invalid")).isEqualTo(PlcResponseCode.INVALID_ADDRESS); + } + + /** + * One bad address must not take the valid tags of the same request down with it. + */ + @Test + void readMixesValidAndInvalidTagAddresses() throws Exception { + when(mockDevice.get(any(SimulatedTag.class))).thenReturn(Optional.of(new PlcSTRING("value"))); + + PlcReadRequest plcReadRequest = SUT.readRequestBuilder() + .addTagAddress("good", "RANDOM/foo:STRING") + .addTagAddress("bad", "RANDOM/foo:Boolean") + .build(); + + PlcReadResponse response = SUT.read(plcReadRequest).get(1, TimeUnit.SECONDS); + + assertThat(response.getResponseCode("good")).isEqualTo(PlcResponseCode.OK); + assertThat(response.getResponseCode("bad")).isEqualTo(PlcResponseCode.INVALID_ADDRESS); + } } @Nested @@ -120,6 +162,20 @@ class SimulatedConnectionTest implements WithAssertions { assertThat(subscriptionHandles).isNotEmpty(); } + /** + * Subscriptions handle a bad address differently from reads and writes: their builder + * rejects it immediately instead of carrying an error item into the request, which is + * why {@code onSubscribe} needs no invalid-address branch. + */ + @Test + void subscribeWithInvalidTagAddressIsRejectedByTheBuilder() { + PlcSubscriptionRequest.Builder builder = SUT.subscriptionRequestBuilder(); + + assertThatThrownBy(() -> builder.addChangeOfStateTagAddress("bad", "STATE/foo:Boolean")) + .isInstanceOf(PlcInvalidTagException.class) + .hasMessageContaining("Boolean"); + } + @Test void unsubscribe() throws Exception { PlcUnsubscriptionRequest plcUnsubscriptionRequest = SUT.unsubscriptionRequestBuilder()
