package com.ruoyi.device.bts;
|
|
import java.io.ByteArrayOutputStream;
|
import java.io.EOFException;
|
import java.io.IOException;
|
import java.io.InputStream;
|
import java.io.OutputStream;
|
import java.net.ConnectException;
|
import java.net.InetSocketAddress;
|
import java.net.Socket;
|
import java.net.SocketTimeoutException;
|
import java.nio.ByteBuffer;
|
import java.nio.charset.CharacterCodingException;
|
import java.nio.charset.CodingErrorAction;
|
import java.nio.charset.StandardCharsets;
|
import java.time.Duration;
|
import java.util.Objects;
|
import java.util.concurrent.locks.ReentrantLock;
|
|
public final class BtsTcpTransport implements BtsTransport
|
{
|
private static final byte[] FRAME_END = "#\r\n".getBytes(StandardCharsets.UTF_8);
|
private static final int DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024;
|
private static final int MAX_RESPONSE_TIMEOUT_MS = 30_000;
|
|
private final String host;
|
private final int port;
|
private final int connectTimeoutMs;
|
private final int responseTimeoutMs;
|
private final int maxResponseBytes;
|
private final ReentrantLock requestLock = new ReentrantLock(true);
|
private Socket socket;
|
|
public BtsTcpTransport(String host, int port, Duration connectTimeout, Duration responseTimeout)
|
{
|
this(host, port, connectTimeout, responseTimeout, DEFAULT_MAX_RESPONSE_BYTES);
|
}
|
|
BtsTcpTransport(String host, int port, Duration connectTimeout, Duration responseTimeout, int maxResponseBytes)
|
{
|
this.host = Objects.requireNonNull(host, "host");
|
this.port = port;
|
this.connectTimeoutMs = Math.toIntExact(connectTimeout.toMillis());
|
this.responseTimeoutMs = Math.toIntExact(responseTimeout.toMillis());
|
this.maxResponseBytes = maxResponseBytes;
|
if (host.isBlank() || port < 1 || port > 65535) throw new IllegalArgumentException("Invalid BTS TCP endpoint");
|
if (connectTimeoutMs < 1 || responseTimeoutMs < 1) throw new IllegalArgumentException("Timeouts must be positive");
|
// 厂商协议声明单条命令最长不超过约 30 秒，避免串行请求队列被错误配置长期阻塞。
|
if (responseTimeoutMs > MAX_RESPONSE_TIMEOUT_MS)
|
throw new IllegalArgumentException("BTS response timeout must not exceed 30 seconds");
|
if (maxResponseBytes < 1) throw new IllegalArgumentException("maxResponseBytes must be positive");
|
}
|
|
@Override
|
public String exchange(String request)
|
{
|
Objects.requireNonNull(request, "request");
|
requestLock.lock();
|
try
|
{
|
ensureConnected();
|
OutputStream output = socket.getOutputStream();
|
output.write(request.getBytes(StandardCharsets.UTF_8));
|
output.write(FRAME_END);
|
output.flush();
|
return readFrame(socket.getInputStream());
|
}
|
catch (ConnectException ex)
|
{
|
closeSocket();
|
throw new BtsTransportException(BtsTransportException.Reason.CONNECTION_FAILED,
|
"BTS TCP connection failed for " + host + ":" + port, ex);
|
}
|
catch (SocketTimeoutException ex)
|
{
|
closeSocket();
|
throw new BtsTransportException(BtsTransportException.Reason.TIMEOUT,
|
"BTS TCP exchange timed out for " + host + ":" + port, ex);
|
}
|
catch (EOFException ex)
|
{
|
closeSocket();
|
throw new BtsTransportException(BtsTransportException.Reason.DISCONNECTED,
|
"BTS disconnected before completing a response for " + host + ":" + port, ex);
|
}
|
catch (ResponseTooLargeException ex)
|
{
|
closeSocket();
|
throw new BtsTransportException(BtsTransportException.Reason.RESPONSE_TOO_LARGE,
|
"BTS response exceeded " + maxResponseBytes + " bytes for " + host + ":" + port, ex);
|
}
|
catch (InvalidUtf8Exception ex)
|
{
|
closeSocket();
|
throw new BtsTransportException(BtsTransportException.Reason.INVALID_UTF8,
|
"BTS response is not valid UTF-8 for " + host + ":" + port, ex);
|
}
|
catch (IOException ex)
|
{
|
closeSocket();
|
throw new BtsTransportException(BtsTransportException.Reason.IO_FAILURE,
|
"BTS TCP exchange failed for " + host + ":" + port, ex);
|
}
|
finally
|
{
|
requestLock.unlock();
|
}
|
}
|
|
private void ensureConnected() throws IOException
|
{
|
if (socket != null && socket.isConnected() && !socket.isClosed()) return;
|
Socket connected = new Socket();
|
connected.connect(new InetSocketAddress(host, port), connectTimeoutMs);
|
connected.setSoTimeout(responseTimeoutMs);
|
socket = connected;
|
}
|
|
private String readFrame(InputStream input) throws IOException
|
{
|
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
|
int matched = 0;
|
while (true)
|
{
|
int value = input.read();
|
if (value < 0) throw new EOFException("BTS closed the connection before the response frame ended");
|
if ((byte) value == FRAME_END[matched])
|
{
|
matched++;
|
if (matched == FRAME_END.length) return decodeUtf8(buffer.toByteArray());
|
continue;
|
}
|
if (matched > 0)
|
{
|
for (int i = 0; i < matched; i++) appendPayload(buffer, FRAME_END[i]);
|
matched = 0;
|
if ((byte) value == FRAME_END[0])
|
{
|
matched = 1;
|
continue;
|
}
|
}
|
appendPayload(buffer, value);
|
}
|
}
|
|
private void appendPayload(ByteArrayOutputStream buffer, int value) throws ResponseTooLargeException
|
{
|
if (buffer.size() >= maxResponseBytes) throw new ResponseTooLargeException();
|
buffer.write(value);
|
}
|
|
private static String decodeUtf8(byte[] bytes) throws InvalidUtf8Exception
|
{
|
try
|
{
|
return StandardCharsets.UTF_8.newDecoder()
|
.onMalformedInput(CodingErrorAction.REPORT)
|
.onUnmappableCharacter(CodingErrorAction.REPORT)
|
.decode(ByteBuffer.wrap(bytes))
|
.toString();
|
}
|
catch (CharacterCodingException ex)
|
{
|
throw new InvalidUtf8Exception(ex);
|
}
|
}
|
|
@Override
|
public void close()
|
{
|
requestLock.lock();
|
try
|
{
|
closeSocket();
|
}
|
finally
|
{
|
requestLock.unlock();
|
}
|
}
|
|
private void closeSocket()
|
{
|
if (socket == null) return;
|
try
|
{
|
socket.close();
|
}
|
catch (IOException ignored)
|
{
|
}
|
finally
|
{
|
socket = null;
|
}
|
}
|
|
private static final class ResponseTooLargeException extends IOException
|
{
|
private static final long serialVersionUID = 1L;
|
}
|
|
private static final class InvalidUtf8Exception extends IOException
|
{
|
private static final long serialVersionUID = 1L;
|
|
private InvalidUtf8Exception(Throwable cause)
|
{
|
super(cause);
|
}
|
}
|
}
|
