package com.battery.device.bts;
|
|
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;
|
// BTS TCP 会话按“单请求、单响应”工作，同一适配器实例不得并发复用连接。
|
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");
|
}
|
|
/**
|
* 在长连接上串行完成一次「单请求、单响应」交换：请求后追加报文结束符 {@code #\r\n}，
|
* 读到同一结束符即视为响应完整。并发调用由锁串行化；任何传输失败都作废当前连接，
|
* 并把失败分类为 {@link BtsTransportException} 抛出，下一次调用重新建连。
|
*/
|
@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 (BtsFrameReader.FrameTooLargeException 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
|
{
|
return decodeUtf8(BtsFrameReader.read(input, maxResponseBytes));
|
}
|
|
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 InvalidUtf8Exception extends IOException
|
{
|
private static final long serialVersionUID = 1L;
|
|
private InvalidUtf8Exception(Throwable cause)
|
{
|
super(cause);
|
}
|
}
|
}
|
