This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new edcddb0e88 fix(plc4j): frame Modbus RTU responses by function code
with resync
edcddb0e88 is described below
commit edcddb0e88f2ba604f14c8b5bf3668051570df3d
Author: Sebastian Rühl <[email protected]>
AuthorDate: Mon Jul 6 21:43:46 2026 +0200
fix(plc4j): frame Modbus RTU responses by function code with resync
The RTU codec treated everything available as a single frame (RTU has no
length field), discarding trailing frames in batched deliveries and
losing partial frames. Responses are now sized by function code
(exceptions 5, reads and file-record ops 5+byteCount, write echoes 8),
validated from a peek by the CRC-checking parser before being consumed,
and unparseable bytes are skipped one at a time until the stream
resynchronizes. This makes multi-slave reuse-port serial setups work
end to end: every frame in a broadcast batch reaches its connection.
---
RELEASE_NOTES | 4 +
.../java/modbus/rtu/ModbusRtuMessageCodec.java | 130 +++++++++++-
.../java/modbus/rtu/ModbusRtuMessageCodecTest.java | 230 ++++++++++++++++++++-
3 files changed, 347 insertions(+), 17 deletions(-)
diff --git a/RELEASE_NOTES b/RELEASE_NOTES
index 3ddb5ed69d..5c18715de6 100644
--- a/RELEASE_NOTES
+++ b/RELEASE_NOTES
@@ -28,6 +28,10 @@ New Features
between connections, and "interframe-delay" write pacing works on both
shared and dedicated ports (gap measured from the last write or
received data).
+- The Java Modbus RTU codec now frames responses by function code with
+ CRC validation and byte-wise resynchronization: batched deliveries
+ (e.g. on shared serial ports) yield every frame instead of only the
+ first, and partial or corrupted frames no longer discard buffered data.
Incompatible changes
--------------------
diff --git
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodec.java
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodec.java
index 2cafb3c631..142ff33249 100644
---
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodec.java
+++
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodec.java
@@ -24,7 +24,9 @@ import org.apache.plc4x.java.modbus.readwrite.ModbusRtuADU;
import org.apache.plc4x.java.spi.buffers.api.exceptions.BufferException;
import org.apache.plc4x.java.spi.buffers.bytebased.ReadBufferByteBased;
import org.apache.plc4x.java.spi.drivers.MessageCodecBase;
+import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException;
import org.apache.plc4x.java.spi.transports.api.TransportInstance;
+import org.apache.plc4x.java.spi.transports.api.exceptions.TransportException;
import java.util.function.Consumer;
@@ -33,13 +35,32 @@ import java.util.function.Consumer;
* Handles the encoding and decoding of Modbus RTU ADU (Application Data Unit)
messages.
*
* RTU framing: 1 byte address + PDU + 2 bytes CRC.
- * Since RTU has no explicit length field, we attempt to parse the available
data
- * and wait for more data if the parse fails.
+ * Since RTU has no explicit length field in the header, the response frame
size is
+ * derived from the function code (see {@link #calculateTotalMessageSize}):
exception
+ * responses (fc >= 0x80) are 5 bytes, read responses (0x01-0x04, 0x14,
0x15, 0x17,
+ * including the file-record operations) are 5 + the byte-count carried in the
third
+ * header byte, and write echoes (0x05, 0x06, 0x0F, 0x10) are a fixed 8 bytes.
+ * {@link #processIncomingData()} peeks
+ * a candidate frame, only consumes it once the generated parser has validated
its CRC,
+ * and otherwise resynchronizes byte-wise.
+ *
+ * Limitation: since the size table above is response-shaped, this codec
assumes it is
+ * only ever used to decode responses (as constructed) - it cannot correctly
frame a
+ * stream of raw Modbus RTU requests.
*/
public class ModbusRtuMessageCodec extends MessageCodecBase<ModbusRtuADU> {
// Minimum RTU message: address (1) + function code (1) + CRC (2) = 4 bytes
private static final int MODBUS_RTU_MIN_SIZE = 4;
+ // address + fc + exception code + CRC
+ private static final int EXCEPTION_RESPONSE_SIZE = 5;
+ // address + fc + 2 bytes address + 2 bytes value/quantity + CRC
+ private static final int WRITE_ECHO_RESPONSE_SIZE = 8;
+ // address + fc + third byte (byteCount for reads) — all the sizing
+ // table needs to compute the expected response size
+ private static final int SIZING_HEADER_SIZE = 3;
+
+ private long resyncSkippedBytes;
public ModbusRtuMessageCodec(TransportInstance<?> transportInstance,
Consumer<ModbusRtuADU> messageHandler) {
super("Modbus RTU", transportInstance, messageHandler);
@@ -50,12 +71,32 @@ public class ModbusRtuMessageCodec extends
MessageCodecBase<ModbusRtuADU> {
return MODBUS_RTU_MIN_SIZE;
}
+ /**
+ * Computes the expected RESPONSE frame size from the peeked header
+ * (address, function code, third byte). RTU has no length field, so
+ * the size follows from the function code:
+ * exception (fc >= 0x80) = 5; reads (0x01-0x04, 0x14, 0x15, 0x17 —
+ * including the file-record read/write operations, whose responses
+ * carry byteCount as the third byte covering the whole item array
+ * just like the plain reads) = 5 + byteCount (the third byte);
+ * write echoes (0x05, 0x06, 0x0F, 0x10) = 8.
+ * Returns -1 for unknown function codes — the caller treats that as
+ * desynchronization, not as "wait for more data".
+ */
@Override
protected int calculateTotalMessageSize(byte[] header, int availableBytes)
{
- // RTU does not have an explicit length field in the header.
- // We return the total available bytes and let parseMessage handle
validation.
- // If the parse fails due to incomplete data, the base class will wait
for more data.
- return availableBytes;
+ int functionCode = header[1] & 0xFF;
+ if (functionCode >= 0x80) {
+ return EXCEPTION_RESPONSE_SIZE;
+ }
+ switch (functionCode) {
+ case 0x01, 0x02, 0x03, 0x04, 0x14, 0x15, 0x17:
+ return MODBUS_RTU_MIN_SIZE + 1 + (header[2] & 0xFF);
+ case 0x05, 0x06, 0x0F, 0x10:
+ return WRITE_ECHO_RESPONSE_SIZE;
+ default:
+ return -1;
+ }
}
@Override
@@ -67,4 +108,81 @@ public class ModbusRtuMessageCodec extends
MessageCodecBase<ModbusRtuADU> {
return modbusADU;
}
+ /**
+ * RTU-specific receive loop: peek before consume. The base class's
+ * loop reads a frame before validating it, which loses bytes whenever
+ * the stream carries more (or less) than exactly one frame — the norm
+ * on a shared serial line. Here a frame is only consumed after the
+ * generated parser (which validates the CRC checksum field) accepted
+ * it from a peek; anything unparseable advances by a single byte
+ * (byte-wise resynchronization, mirroring the plc4go modbus codec).
+ * <p>
+ * While resynchronizing (i.e. a prior candidate at this stream position
+ * already failed), a header that looks valid but demands more bytes than
+ * are available is treated as another failed candidate rather than as a
+ * reason to wait: a byte-shifted resync candidate can accidentally look
+ * like a legitimate function code with a large byte count, which would
+ * otherwise stall the codec forever waiting for bytes that were never
+ * going to arrive. Waiting is only correct at a clean (non-resyncing)
+ * stream position, where an undersized buffer really does mean "the next
+ * frame hasn't fully arrived yet". The corollary trade-off: a GENUINE
+ * partial frame sitting directly behind garbage is skipped rather than
+ * awaited — that response is lost and recovered by the master's own
+ * request timeout and retry, which beats stalling every frame queued
+ * behind an undersized garbage candidate that will never complete.
+ */
+ @Override
+ public void processIncomingData() throws MessageCodecException {
+ try {
+ while (true) {
+ int availableBytes =
getTransportInstance().getNumBytesAvailable();
+ if (availableBytes < MODBUS_RTU_MIN_SIZE) {
+ return;
+ }
+ byte[] header =
getTransportInstance().peekReadableBytes(SIZING_HEADER_SIZE);
+ int expectedSize = calculateTotalMessageSize(header,
availableBytes);
+ if (expectedSize < 0) {
+ skipOneByte("unknown function code 0x" +
Integer.toHexString(header[1] & 0xFF));
+ continue;
+ }
+ if (availableBytes < expectedSize) {
+ if (resyncSkippedBytes > 0) {
+ skipOneByte("candidate frame during resync needs " +
expectedSize
+ + " bytes, only " + availableBytes + " available");
+ continue;
+ }
+ return; // never consume a partial frame
+ }
+ byte[] frame =
getTransportInstance().peekReadableBytes(expectedSize);
+ ModbusRtuADU message;
+ try {
+ message = parseMessage(createReadBuffer(frame));
+ } catch (BufferException e) {
+ skipOneByte("frame failed validation: " + e.getMessage());
+ continue;
+ }
+ getTransportInstance().read(expectedSize); // consume the
validated frame
+ noteResyncComplete();
+ messageHandler.accept(message);
+ }
+ } catch (TransportException e) {
+ throw new MessageCodecException("Failed to receive Modbus RTU
message", e);
+ }
+ }
+
+ private void skipOneByte(String reason) throws TransportException {
+ if (resyncSkippedBytes == 0) {
+ logger.warn("Modbus RTU stream out of sync ({}), resynchronizing
byte-wise", reason);
+ }
+ getTransportInstance().read(1);
+ resyncSkippedBytes++;
+ }
+
+ private void noteResyncComplete() {
+ if (resyncSkippedBytes > 0) {
+ logger.warn("Modbus RTU stream resynchronized after skipping {}
bytes", resyncSkippedBytes);
+ resyncSkippedBytes = 0;
+ }
+ }
+
}
diff --git
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodecTest.java
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodecTest.java
index b49b66ae20..ddb322332f 100644
---
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodecTest.java
+++
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuMessageCodecTest.java
@@ -18,11 +18,17 @@
*/
package org.apache.plc4x.java.modbus.rtu;
-import org.apache.plc4x.java.modbus.readwrite.ModbusRtuADU;
+import org.apache.plc4x.java.modbus.readwrite.*;
+import org.apache.plc4x.java.spi.buffers.bytebased.WriteBufferByteBased;
import org.apache.plc4x.java.spi.transports.api.TransportInstance;
+import org.apache.plc4x.java.spi.transports.api.config.TransportConfiguration;
+import org.apache.plc4x.java.spi.transports.api.exceptions.TransportException;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
import java.util.function.Consumer;
import static org.junit.jupiter.api.Assertions.*;
@@ -52,18 +58,220 @@ class ModbusRtuMessageCodecTest {
assertEquals(4, method.invoke(codec));
}
- @SuppressWarnings("unchecked")
+ /**
+ * In-memory transport double: bytes are queued via deliver(...) and the
+ * codec consumes them through the real TransportInstance contract.
+ * Strict: peeking/reading more than available is a test bug and throws.
+ */
+ static final class ScriptedTransport implements
TransportInstance<TransportConfiguration> {
+ private final ByteArrayOutputStream buffer = new
ByteArrayOutputStream();
+ private int readPosition;
+ private boolean open = true;
+
+ void deliver(byte[] bytes) {
+ buffer.writeBytes(bytes);
+ }
+
+ private byte[] bufferedBytes() {
+ return buffer.toByteArray();
+ }
+
+ @Override
+ public TransportConfiguration getConfiguration() {
+ return null;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public int getNumBytesAvailable() {
+ return buffer.size() - readPosition;
+ }
+
+ @Override
+ public byte[] peekReadableBytes(int numBytes) throws
TransportException {
+ if (numBytes > getNumBytesAvailable()) {
+ throw new TransportException("peek beyond available: " +
numBytes);
+ }
+ byte[] all = bufferedBytes();
+ byte[] result = new byte[numBytes];
+ System.arraycopy(all, readPosition, result, 0, numBytes);
+ return result;
+ }
+
+ @Override
+ public byte[] read(int numBytes) throws TransportException {
+ byte[] result = peekReadableBytes(numBytes);
+ readPosition += numBytes;
+ return result;
+ }
+
+ @Override
+ public void write(byte[] bytes) {
+ // not exercised by these tests
+ }
+
+ @Override
+ public void close() {
+ open = false;
+ }
+ }
+
+ private static byte[] wireBytes(ModbusRtuADU adu) throws Exception {
+ WriteBufferByteBased writeBuffer = new WriteBufferByteBased(new
byte[adu.getLengthInBytes()]);
+ adu.serialize(writeBuffer);
+ return writeBuffer.getBytes();
+ }
+
+ private static byte[] readResponseFrame(int unitId, byte[] registers)
throws Exception {
+ return wireBytes(new ModbusRtuADU((short) unitId, new
ModbusPDUReadHoldingRegistersResponse(registers)));
+ }
+
+ private static byte[] readFileRecordResponseFrame(int unitId, byte[] data)
throws Exception {
+ List<ModbusPDUReadFileRecordResponseItem> items = new ArrayList<>();
+ items.add(new ModbusPDUReadFileRecordResponseItem((short) 6, data));
+ return wireBytes(new ModbusRtuADU((short) unitId, new
ModbusPDUReadFileRecordResponse(items)));
+ }
+
+ private static byte[] concat(byte[]... parts) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ for (byte[] part : parts) {
+ out.writeBytes(part);
+ }
+ return out.toByteArray();
+ }
+
@Test
- void testCalculateTotalMessageSize_returnsAvailableBytes() throws
Exception {
- TransportInstance<?> transportInstance = mock(TransportInstance.class);
- Consumer<ModbusRtuADU> handler = mock(Consumer.class);
- ModbusRtuMessageCodec codec = new
ModbusRtuMessageCodec(transportInstance, handler);
+ void singleCompleteResponseIsDispatched() throws Exception {
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
- Method method =
codec.getClass().getDeclaredMethod("calculateTotalMessageSize", byte[].class,
int.class);
- method.setAccessible(true);
+ transport.deliver(readResponseFrame(1, new byte[]{0x11, 0x22}));
+ codec.processIncomingData();
+
+ assertEquals(1, received.size());
+ assertEquals(1, received.get(0).getAddress());
+ }
+
+ @Test
+ void twoResponsesInOneDeliveryAreBothDispatched() throws Exception {
+ // The shared-port batching case — the bug this change fixes.
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
+
+ transport.deliver(concat(
+ readResponseFrame(1, new byte[]{0x11, 0x22}),
+ readResponseFrame(2, new byte[]{0x33, 0x44, 0x55, 0x66})));
+ codec.processIncomingData();
+
+ assertEquals(2, received.size());
+ assertEquals(1, received.get(0).getAddress());
+ assertEquals(2, received.get(1).getAddress());
+ }
+
+ @Test
+ void partialFrameWaitsAndCompletesAcrossDeliveries() throws Exception {
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
+
+ byte[] frame = readResponseFrame(1, new byte[]{0x11, 0x22, 0x33,
0x44});
+ transport.deliver(java.util.Arrays.copyOfRange(frame, 0, 5)); //
header + partial data
+ codec.processIncomingData();
+ assertEquals(0, received.size(), "nothing may be consumed while the
frame is incomplete");
+ assertEquals(5, transport.getNumBytesAvailable(), "partial bytes must
remain buffered");
+
+ transport.deliver(java.util.Arrays.copyOfRange(frame, 5,
frame.length));
+ codec.processIncomingData();
+ assertEquals(1, received.size());
+ }
+
+ @Test
+ void garbagePrefixIsSkippedToTheNextValidFrame() throws Exception {
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
+
+ // 0x00 is not a known function code in byte 2 position for any
+ // alignment of this prefix — forces byte-wise resync.
+ transport.deliver(concat(new byte[]{0x00, 0x00, 0x00},
readResponseFrame(3, new byte[]{0x77, 0x77})));
+ codec.processIncomingData();
+
+ assertEquals(1, received.size());
+ assertEquals(3, received.get(0).getAddress());
+ }
+
+ @Test
+ void corruptCrcFrameIsSkippedAndNextFrameParses() throws Exception {
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
+
+ byte[] corrupt = readResponseFrame(1, new byte[]{0x11, 0x22});
+ corrupt[corrupt.length - 1] ^= (byte) 0xFF; // break the CRC
+ transport.deliver(concat(corrupt, readResponseFrame(2, new
byte[]{0x33, 0x44})));
+ codec.processIncomingData();
+
+ assertEquals(1, received.size(), "the corrupt frame is skipped, the
valid one parses");
+ assertEquals(2, received.get(0).getAddress());
+ }
+
+ @Test
+ void writeEchoAndExceptionResponsesAreFramedCorrectly() throws Exception {
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
+
+ byte[] writeEcho = wireBytes(new ModbusRtuADU((short) 4, new
ModbusPDUWriteSingleRegisterResponse(0x0010, 0x1234)));
+ assertEquals(8, writeEcho.length, "sanity: write echo is the fixed
8-byte frame");
+ byte[] exception = wireBytes(new ModbusRtuADU((short) 5, new
ModbusPDUError(ModbusErrorCode.ILLEGAL_FUNCTION)));
+ assertEquals(5, exception.length, "sanity: exception response is the
fixed 5-byte frame");
+
+ transport.deliver(concat(writeEcho, exception));
+ codec.processIncomingData();
+
+ assertEquals(2, received.size());
+ assertEquals(4, received.get(0).getAddress());
+ assertEquals(5, received.get(1).getAddress());
+ }
+
+ @Test
+ void fileRecordResponseIsFramedCorrectly() throws Exception {
+ // fc 0x14 (read file record) responses carry byteCount as the third
+ // byte just like the plain reads, but the sizing switch used to
+ // omit 0x14/0x15 entirely, so this frame would be discarded byte-wise
+ // and the following frame would be the only one dispatched.
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
+
+ transport.deliver(concat(
+ readFileRecordResponseFrame(1, new byte[]{0x11, 0x22, 0x33, 0x44}),
+ readResponseFrame(2, new byte[]{0x33, 0x44})));
+ codec.processIncomingData();
+
+ assertEquals(2, received.size(), "fc 0x14 (read file record) response
must be framed, not discarded");
+ assertEquals(1, received.get(0).getAddress());
+ assertEquals(2, received.get(1).getAddress());
+ }
+
+ @Test
+ void foreignUnitIdFrameIsStillDispatched() throws Exception {
+ // The codec frames and dispatches everything on the wire; unit-id
+ // filtering is the connection layer's job.
+ ScriptedTransport transport = new ScriptedTransport();
+ List<ModbusRtuADU> received = new ArrayList<>();
+ ModbusRtuMessageCodec codec = new ModbusRtuMessageCodec(transport,
received::add);
+
+ transport.deliver(readResponseFrame(42, new byte[]{0x01, 0x02}));
+ codec.processIncomingData();
- // RTU has no explicit length, so it returns the available bytes
- byte[] header = new byte[]{0x01, 0x03, 0x00, 0x00};
- assertEquals(20, method.invoke(codec, header, 20));
+ assertEquals(1, received.size());
+ assertEquals(42, received.get(0).getAddress());
}
}
