Merge PR #19: feat(modbus): add TcpTransport with MBAP framing and a mock server
This commit is contained in:
root committed
commit
5d3201f7e3
3 files changed
+381
No files matched your search
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<Integer, Integer> holding = new ConcurrentHashMap<>();
|
||||
private final Map<Integer, Integer> 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<Integer, Integer> 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();
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user