diff --git a/src/main/java/home/gateway/adapters/modbus/TcpTransport.java b/src/main/java/home/gateway/adapters/modbus/TcpTransport.java new file mode 100644 index 0000000..ed04d83 --- /dev/null +++ b/src/main/java/home/gateway/adapters/modbus/TcpTransport.java @@ -0,0 +1,133 @@ +package home.gateway.adapters.modbus; + +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.time.Duration; + +/** + * Modbus TCP (MBAP) over one long-lived socket. The MK01 gateway serves a single client and a + * second connection drops the first, so the socket is reused and requests never overlap. + */ +public final class TcpTransport implements ModbusTransport { + + public record Settings(String host, int port, Duration connectTimeout, Duration readTimeout, + Duration minimumPause) { + } + + private static final int HEADER_LENGTH = 7; + + private final Settings settings; + private Socket socket; + private DataInputStream in; + private int transactionId; + private long lastResponseNanos; + + public TcpTransport(Settings settings) { + this.settings = settings; + this.lastResponseNanos = System.nanoTime() - settings.minimumPause().toNanos(); + } + + @Override + public synchronized byte[] exchange(int unitId, byte[] requestPdu) throws IOException { + try { + awaitPause(); + connectIfNeeded(); + int id = nextTransactionId(); + socket.getOutputStream().write(frame(id, unitId, requestPdu)); + byte[] response = readResponse(id, unitId); + lastResponseNanos = System.nanoTime(); + return response; + } catch (IOException e) { + closeSocket(); + throw e; + } + } + + @Override + public synchronized void close() { + closeSocket(); + } + + private void awaitPause() throws InterruptedIOException { + long remaining = settings.minimumPause().toNanos() - (System.nanoTime() - lastResponseNanos); + if (remaining <= 0) { + return; + } + try { + Thread.sleep(Duration.ofNanos(remaining)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("interrupted while waiting between requests"); + } + } + + private void connectIfNeeded() throws IOException { + if (socket != null) { + return; + } + Socket candidate = new Socket(); + try { + candidate.setTcpNoDelay(true); + candidate.setSoTimeout((int) settings.readTimeout().toMillis()); + candidate.connect(new InetSocketAddress(settings.host(), settings.port()), + (int) settings.connectTimeout().toMillis()); + } catch (IOException e) { + candidate.close(); + throw e; + } + socket = candidate; + in = new DataInputStream(candidate.getInputStream()); + } + + private int nextTransactionId() { + transactionId = (transactionId + 1) & 0xFFFF; + return transactionId; + } + + private static byte[] frame(int transactionId, int unitId, byte[] pdu) { + return ByteBuffer.allocate(HEADER_LENGTH + pdu.length) + .putShort((short) transactionId) + .putShort((short) 0) + .putShort((short) (pdu.length + 1)) + .put((byte) unitId) + .put(pdu) + .array(); + } + + private byte[] readResponse(int expectedTransactionId, int expectedUnitId) throws IOException { + ByteBuffer header = ByteBuffer.wrap(in.readNBytes(HEADER_LENGTH)); + if (header.remaining() < HEADER_LENGTH) { + throw new IOException("connection closed by peer"); + } + int id = header.getShort() & 0xFFFF; + int protocol = header.getShort() & 0xFFFF; + int length = header.getShort() & 0xFFFF; + int unitId = header.get() & 0xFF; + if (protocol != 0 || id != expectedTransactionId || unitId != expectedUnitId || length < 2) { + throw new IOException("unexpected MBAP header: transaction %d, protocol %d, length %d, unit %d" + .formatted(id, protocol, length, unitId)); + } + byte[] pdu = in.readNBytes(length - 1); + if (pdu.length < length - 1) { + throw new IOException("connection closed by peer"); + } + return pdu; + } + + private void closeSocket() { + if (socket == null) { + return; + } + try { + socket.close(); + } catch (IOException ignored) { + // nothing useful to do with a socket that fails to close + } + socket = null; + in = null; + } +} diff --git a/src/test/java/home/gateway/adapters/modbus/MockModbusServer.java b/src/test/java/home/gateway/adapters/modbus/MockModbusServer.java new file mode 100644 index 0000000..92993c1 --- /dev/null +++ b/src/test/java/home/gateway/adapters/modbus/MockModbusServer.java @@ -0,0 +1,145 @@ +package home.gateway.adapters.modbus; + +import java.io.Closeable; +import java.io.DataInputStream; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** Local Modbus TCP fake answering FC3 and FC4; unknown addresses get exception code 2. */ +final class MockModbusServer implements Closeable { + + private final ServerSocket server; + private final Map holding = new ConcurrentHashMap<>(); + private final Map input = new ConcurrentHashMap<>(); + private final AtomicInteger connections = new AtomicInteger(); + private volatile Duration responseDelay = Duration.ZERO; + private volatile boolean dropNextRequest; + private volatile boolean wrongTransactionId; + + MockModbusServer() throws IOException { + server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread.ofVirtual().start(this::acceptLoop); + } + + int port() { + return server.getLocalPort(); + } + + int connections() { + return connections.get(); + } + + void holding(int address, int value) { + holding.put(address, value); + } + + void input(int address, int value) { + input.put(address, value); + } + + void delayResponses(Duration delay) { + responseDelay = delay; + } + + /** Closes the connection on the next request without answering it. */ + void dropNextRequest() { + dropNextRequest = true; + } + + /** Answers the next request with a transaction id that does not match. */ + void answerWithWrongTransactionId() { + wrongTransactionId = true; + } + + @Override + public void close() throws IOException { + server.close(); + } + + private void acceptLoop() { + while (!server.isClosed()) { + try { + Socket client = server.accept(); + connections.incrementAndGet(); + Thread.ofVirtual().start(() -> serve(client)); + } catch (IOException e) { + return; + } + } + } + + private void serve(Socket client) { + try (client) { + DataInputStream in = new DataInputStream(client.getInputStream()); + while (true) { + ByteBuffer header = ByteBuffer.wrap(in.readNBytes(7)); + if (header.remaining() < 7) { + return; + } + int transactionId = header.getShort() & 0xFFFF; + header.getShort(); + int length = header.getShort() & 0xFFFF; + int unitId = header.get() & 0xFF; + byte[] pdu = in.readNBytes(length - 1); + if (dropNextRequest) { + dropNextRequest = false; + return; + } + pause(); + if (wrongTransactionId) { + wrongTransactionId = false; + transactionId++; + } + client.getOutputStream().write(frame(transactionId, unitId, answer(pdu))); + } + } catch (IOException ignored) { + // the client went away, which is a normal end of a test + } + } + + private void pause() { + try { + Thread.sleep(responseDelay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private byte[] answer(byte[] pdu) { + int function = pdu[0]; + Map registers = switch (function) { + case 3 -> holding; + case 4 -> input; + default -> null; + }; + if (registers == null) { + return new byte[] {(byte) (function | 0x80), 1}; + } + ByteBuffer request = ByteBuffer.wrap(pdu, 1, 4); + int address = request.getShort() & 0xFFFF; + int count = request.getShort() & 0xFFFF; + ByteBuffer response = ByteBuffer.allocate(2 + 2 * count).put((byte) function).put((byte) (2 * count)); + for (int i = 0; i < count; i++) { + Integer value = registers.get(address + i); + if (value == null) { + return new byte[] {(byte) (function | 0x80), 2}; + } + response.putShort(value.shortValue()); + } + return response.array(); + } + + private static byte[] frame(int transactionId, int unitId, byte[] pdu) { + return ByteBuffer.allocate(7 + pdu.length) + .putShort((short) transactionId).putShort((short) 0) + .putShort((short) (pdu.length + 1)).put((byte) unitId).put(pdu) + .array(); + } +} diff --git a/src/test/java/home/gateway/adapters/modbus/TcpTransportTest.java b/src/test/java/home/gateway/adapters/modbus/TcpTransportTest.java new file mode 100644 index 0000000..71a2465 --- /dev/null +++ b/src/test/java/home/gateway/adapters/modbus/TcpTransportTest.java @@ -0,0 +1,103 @@ +package home.gateway.adapters.modbus; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.net.SocketTimeoutException; +import java.time.Duration; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class TcpTransportTest { + + private static final byte[] READ_HOLDING_10 = {3, 0, 10, 0, 1}; + private static final byte[] HOLDING_10_RESPONSE = {3, 2, 0x12, 0x34}; + + private MockModbusServer server; + private TcpTransport transport; + + @BeforeEach + void start() throws IOException { + server = new MockModbusServer(); + server.holding(10, 0x1234); + server.input(20, 0xFFFE); + transport = transportWith(Duration.ofSeconds(2), Duration.ZERO); + } + + @AfterEach + void stop() throws IOException { + transport.close(); + server.close(); + } + + private TcpTransport transportWith(Duration readTimeout, Duration pause) { + return new TcpTransport(new TcpTransport.Settings( + "localhost", server.port(), Duration.ofSeconds(2), readTimeout, pause)); + } + + @Test + void roundTripsHoldingAndInputReads() throws IOException { + assertArrayEquals(HOLDING_10_RESPONSE, transport.exchange(1, READ_HOLDING_10)); + assertArrayEquals(new byte[] {4, 2, (byte) 0xFF, (byte) 0xFE}, + transport.exchange(1, new byte[] {4, 0, 20, 0, 1})); + } + + @Test + void returnsExceptionPduAsIs() throws IOException { + assertArrayEquals(new byte[] {(byte) 0x83, 2}, transport.exchange(1, new byte[] {3, 0, 99, 0, 1})); + } + + @Test + void reusesOneConnection() throws IOException { + for (int i = 0; i < 20; i++) { + transport.exchange(1, READ_HOLDING_10); + } + assertEquals(1, server.connections()); + } + + @Test + void reconnectsAfterServerDropsConnection() throws IOException { + transport.exchange(1, READ_HOLDING_10); + server.dropNextRequest(); + assertThrows(IOException.class, () -> transport.exchange(1, READ_HOLDING_10)); + + assertArrayEquals(HOLDING_10_RESPONSE, transport.exchange(1, READ_HOLDING_10)); + assertEquals(2, server.connections()); + } + + @Test + void readTimeoutClosesSocketAndLaterExchangeReconnects() throws IOException { + try (TcpTransport impatient = transportWith(Duration.ofMillis(100), Duration.ZERO)) { + server.delayResponses(Duration.ofMillis(500)); + assertThrows(SocketTimeoutException.class, () -> impatient.exchange(1, READ_HOLDING_10)); + + server.delayResponses(Duration.ZERO); + assertArrayEquals(HOLDING_10_RESPONSE, impatient.exchange(1, READ_HOLDING_10)); + assertEquals(2, server.connections()); + } + } + + @Test + void waitsMinimumPauseBetweenRequests() throws IOException { + try (TcpTransport paced = transportWith(Duration.ofSeconds(2), Duration.ofMillis(200))) { + paced.exchange(1, READ_HOLDING_10); + long start = System.nanoTime(); + paced.exchange(1, READ_HOLDING_10); + Duration waited = Duration.ofNanos(System.nanoTime() - start); + assertTrue(waited.toMillis() >= 180, "waited only " + waited); + } + } + + @Test + void rejectsMismatchedTransactionIdAndReconnectsAfterwards() throws IOException { + server.answerWithWrongTransactionId(); + assertThrows(IOException.class, () -> transport.exchange(1, READ_HOLDING_10)); + + assertArrayEquals(HOLDING_10_RESPONSE, transport.exchange(1, READ_HOLDING_10)); + assertEquals(2, server.connections()); + } +}