package com.ard.work.sdk.zlxd.netty.tcp;
|
|
import com.ard.common.core.constant.CacheConstants;
|
import com.ard.common.redis.service.RedisService;
|
import com.ard.work.api.domian.ArdCamera;
|
import com.ard.work.event.LoginEvent;
|
import com.ard.work.sdk.zlxd.netty.message.response.LoginResponse;
|
import com.ard.work.sdk.zlxd.service.ErrorCodeEnum;
|
import io.netty.buffer.ByteBuf;
|
import io.netty.buffer.Unpooled;
|
import io.netty.channel.ChannelHandlerContext;
|
import io.netty.channel.SimpleChannelInboundHandler;
|
import io.netty.handler.timeout.IdleState;
|
import io.netty.handler.timeout.IdleStateEvent;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang3.StringUtils;
|
import org.springframework.context.ApplicationEventPublisher;
|
|
import java.nio.charset.StandardCharsets;
|
import java.util.regex.Matcher;
|
import java.util.regex.Pattern;
|
|
/**
|
* 自定义HTTP协议处理器（直接处理原始ByteBuf，无任何Netty HTTP编解码器依赖）
|
* 适配设备双向HTTP报文：
|
* 1. 设备主动发：POST * HTTP/1.1（业务请求/Heart_Beat心跳，Content-Type可text/html/application/xml）
|
* 2. 设备响应发：HTTP/1.1 200 OK（客户端请求的应答）
|
* 3. 序列号格式：CSeq: X POST
|
* 4. 基于Content-Length+\\r\\n\\r\\n处理TCP粘包拆包
|
*/
|
@Slf4j
|
public class CustomHttpChannelHandler extends SimpleChannelInboundHandler<ByteBuf> {
|
private final ApplicationEventPublisher eventPublisher;
|
|
// 相机唯一ID（区分多相机）
|
private final String cameraId;
|
// TCP粘包拆包核心：缓存未解析的字节流
|
private final ByteBuf cacheBuf = Unpooled.buffer();
|
// 解析CSeq: X POST 中的数字X（正则匹配）
|
private static final Pattern CSEQ_PATTERN = Pattern.compile("CSeq:\\s*(\\d+)\\s*POST", Pattern.CASE_INSENSITIVE);
|
// 心跳标识：XML中的Message_Type="Heart_Beat"
|
private static final String HEART_BEAT_FLAG = "Message_Type=\"Heart_Beat\"";
|
// 心跳应答XML模板（匹配设备格式，填充序列号/会话ID）
|
private static final String HEART_BEAT_RSP_TPL = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?>" +
|
"<Message Version=\"1.0\">" +
|
" <Header Message_Type=\"Heart_Beat_Rsp\" Sequence_Number=\"%s\" Session_ID=\"%s\" Source_ID=\"\" " +
|
"Destination_ID=\"\" />" +
|
" <result code=\"0\" />" +
|
"</Message>";
|
|
public CustomHttpChannelHandler(String cameraId, ApplicationEventPublisher eventPublisher) {
|
this.cameraId = cameraId;
|
this.eventPublisher = eventPublisher;
|
}
|
|
// ===================== 核心方法：处理所有原始ByteBuf字节流 =====================
|
@Override
|
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
|
// 1. 新字节流写入缓存，解决TCP粘包拆包
|
cacheBuf.writeBytes(msg);
|
// 2. 循环解析缓存，直到无完整HTTP包（粘包时一次解析多个，拆包时等待下一次）
|
while (parseOneCompleteHttpPackage(ctx)) {
|
// 空循环，核心逻辑在parseOneCompleteHttpPackage中
|
}
|
}
|
|
// ===================== 关键方法：解析单个完整的HTTP包（核心逻辑）=====================
|
private boolean parseOneCompleteHttpPackage(ChannelHandlerContext ctx) {
|
// 1. 查找HTTP头和体的分隔符：\r\n\r\n（必须找到才是完整包的前提）
|
int delimiterIndex = findDelimiterIndex(cacheBuf, "\r\n\r\n".getBytes(StandardCharsets.UTF_8));
|
if (delimiterIndex == -1) {
|
return false; // 无分隔符，等待下一次字节流
|
}
|
|
try {
|
// 2. 读取并解析HTTP头（0 -> 分隔符索引）
|
ByteBuf headerBuf = cacheBuf.readRetainedSlice(delimiterIndex);
|
String headerStr = headerBuf.toString(StandardCharsets.UTF_8).trim();
|
headerBuf.release();
|
cacheBuf.skipBytes(4); // 跳过分隔符\r\n\r\n（4个字节）
|
|
// 3. 解析Content-Length（必须有，否则无法获取体长度）
|
int contentLength = parseContentLength(headerStr);
|
if (contentLength <= 0) {
|
log.warn("【相机{}】解析HTTP包失败：Content-Length无效 | 头内容:{}", cameraId, headerStr);
|
return true;
|
}
|
|
// 4. 校验缓存体字节是否足够（拆包场景，等待下一次字节流）
|
if (cacheBuf.readableBytes() < contentLength) {
|
cacheBuf.readerIndex(0); // 重置读指针，等待后续字节流
|
return false;
|
}
|
|
// 5. 读取HTTP体（严格按Content-Length读取，避免粘包）
|
ByteBuf bodyBuf = cacheBuf.readRetainedSlice(contentLength);
|
String bodyStr = bodyBuf.toString(StandardCharsets.UTF_8).trim();
|
bodyBuf.release();
|
|
// 6. 按首行区分【设备请求】和【设备响应】（核心区分逻辑，匹配你的报文）
|
String firstLine = headerStr.split("\r\n")[0].trim();
|
if (firstLine.startsWith("POST")) {
|
// 设备主动发的请求（心跳/主动业务请求）→ 处理心跳/请求
|
handleDeviceRequest(ctx, headerStr, bodyStr);
|
} else if (firstLine.startsWith("HTTP/1.1")) {
|
// 设备对客户端的响应（GetPtzInfo/GetDeviceState）→ 完成Future回调
|
handleDeviceResponse(ctx, headerStr, bodyStr);
|
} else {
|
log.warn("【相机{}】解析到未知HTTP包，忽略 | 首行:{}", cameraId, firstLine);
|
}
|
return true;
|
} catch (Exception e) {
|
log.error("【相机{}】解析HTTP包异常，重置缓存", cameraId, e);
|
cacheBuf.readerIndex(0); // 解析异常重置指针，避免后续解析全部失效
|
return false;
|
}
|
}
|
|
// ===================== 处理设备主动请求（心跳/主动业务请求）=====================
|
private void handleDeviceRequest(ChannelHandlerContext ctx, String headerStr, String bodyStr) {
|
// 解析核心参数：CSeq、Session_ID、Sequence_Number、Content-Type
|
String cseqHeader = parseHeaderValue(headerStr, "CSeq");
|
String sessionId = parseXmlAttr(bodyStr, "Session_ID=\"");
|
String seqNum = parseXmlAttr(bodyStr, "Sequence_Number=\"");
|
String contentType = parseHeaderValue(headerStr, "Content-Type");
|
|
//log.debug("【相机{}】收到设备主动请求 | 首行:{} | CSeq:{} | Content-Type:{} | 体长度:{}",
|
// cameraId, headerStr.split("\r\n")[0].trim(), cseqHeader, contentType, bodyStr.length());
|
|
// 心跳识别&自动应答（核心：匹配XML中的Heart_Beat标识）
|
if (bodyStr.contains(HEART_BEAT_FLAG)) {
|
//log.debug("【相机{}】处理设备心跳 | CSeq:{} | Session_ID:{} | 开始发送应答",
|
//cameraId, cseqHeader, sessionId);
|
// 填充心跳应答XML（匹配设备格式，序列号/会话ID和请求一致）
|
String rspXml = String.format(HEART_BEAT_RSP_TPL,
|
StringUtils.isBlank(seqNum) ? "0" : seqNum,
|
StringUtils.isBlank(sessionId) ? "" : sessionId);
|
// 手动拼接HTTP应答并发送
|
sendHttpResponse(ctx, cseqHeader, rspXml);
|
//log.debug("【相机{}】心跳应答发送成功 | CSeq:{} | 应答体长度:{}", cameraId, cseqHeader, rspXml.length());
|
return;
|
}
|
|
// 扩展：处理设备其他主动业务请求（如有，在这里添加逻辑）
|
log.info("相机【{}】收到设备主动业务请求 | 体内容:{}", cameraId, bodyStr);
|
}
|
|
// ===================== 处理设备响应（客户端请求的返回，完成Future回调）=====================
|
private void handleDeviceResponse(ChannelHandlerContext ctx, String headerStr, String bodyStr) {
|
// 1. 解析CSeq序列号（优先从HTTP头解析，兼容XML解析）
|
Integer cseq = parseCSeq(headerStr);
|
if (cseq == null) {
|
cseq = parseSeqFromXml(bodyStr);
|
if (cseq == null) {
|
log.debug("【相机{}】收到设备响应，无有效CSeq，忽略 | 体内容:{}", cameraId, bodyStr);
|
return;
|
}
|
//log.debug("【相机{}】响应头无有效CSeq，从XML解析到:{}", cameraId, cseq);
|
}
|
|
// 2. 空体校验（无效响应直接忽略）
|
if (StringUtils.isBlank(bodyStr)) {
|
log.debug("【相机{}】收到设备响应，体为空，忽略 | CSeq:{}", cameraId, cseq);
|
return;
|
}
|
|
// 3. 完成Future回调（核心：和sendXmlAndWait中的唯一键匹配）
|
String uniqueKey = cameraId + "_" + cseq;
|
//log.debug("【相机{}】收到设备正常响应 | CSeq:{} | 体长度:{} | 完成Future回调", cameraId, cseq, bodyStr.length());
|
ResponseFutureHolder.complete(uniqueKey, bodyStr);
|
}
|
|
// ===================== 手动拼接HTTP应答并发送（心跳/业务应答通用）=====================
|
private void sendHttpResponse(ChannelHandlerContext ctx, String cseqHeader, String bodyStr) {
|
try {
|
byte[] bodyBytes = bodyStr.getBytes(StandardCharsets.UTF_8);
|
// 拼接HTTP应答头（严格匹配设备的响应格式：HTTP/1.1 200 OK、CSeq一致）
|
StringBuilder rspHeader = new StringBuilder();
|
rspHeader.append("HTTP/1.1 200 OK\r\n");
|
if (StringUtils.isNotBlank(cseqHeader)) {
|
rspHeader.append("CSeq: ").append(cseqHeader).append("\r\n");
|
}
|
rspHeader.append("Content-Length: ").append(bodyBytes.length).append("\r\n");
|
rspHeader.append("Content-Type: application/xml; charset=UTF-8\r\n");
|
rspHeader.append("\r\n"); // 头和体的分隔符
|
|
// 拼接完整应答字节流（头+体），发送原始ByteBuf
|
byte[] rspBytes = (rspHeader.toString() + bodyStr).getBytes(StandardCharsets.UTF_8);
|
ByteBuf rspBuf = Unpooled.copiedBuffer(rspBytes);
|
ctx.writeAndFlush(rspBuf).addListener(f -> {
|
if (!f.isSuccess()) {
|
log.error("【相机{}】发送HTTP应答失败 | CSeq:{}", cameraId, cseqHeader, f.cause());
|
}
|
});
|
} catch (Exception e) {
|
log.error("【相机{}】拼接/发送HTTP应答异常", cameraId, e);
|
}
|
}
|
|
// ===================== 工具方法：解析HTTP头中的Content-Length =====================
|
private int parseContentLength(String headerStr) {
|
String[] headerLines = headerStr.split("\r\n");
|
for (String line : headerLines) {
|
if (line.trim().toLowerCase().startsWith("content-length:")) {
|
try {
|
return Integer.parseInt(line.split(":", 2)[1].trim());
|
} catch (Exception e) {
|
log.warn("【相机{}】解析Content-Length失败 | 行:{}", cameraId, line);
|
}
|
}
|
}
|
return -1;
|
}
|
|
// ===================== 工具方法：解析HTTP头中的指定字段值（大小写不敏感）=====================
|
private String parseHeaderValue(String headerStr, String headerKey) {
|
String[] headerLines = headerStr.split("\r\n");
|
String lowerKey = headerKey.toLowerCase() + ":";
|
for (String line : headerLines) {
|
if (line.trim().toLowerCase().startsWith(lowerKey)) {
|
return line.split(":", 2)[1].trim();
|
}
|
}
|
return null;
|
}
|
|
// ===================== 工具方法：解析CSeq: X POST 中的数字X =====================
|
private Integer parseCSeq(String headerStr) {
|
String cseqHeader = parseHeaderValue(headerStr, "CSeq");
|
if (StringUtils.isBlank(cseqHeader)) {
|
return null;
|
}
|
Matcher matcher = CSEQ_PATTERN.matcher(cseqHeader);
|
if (matcher.find()) {
|
try {
|
return Integer.parseInt(matcher.group(1));
|
} catch (Exception e) {
|
log.warn("【相机{}】解析CSeq失败 | 值:{}", cameraId, cseqHeader);
|
}
|
}
|
return null;
|
}
|
|
// ===================== 工具方法：从XML中解析指定属性值（如Session_ID="123" → 123）=====================
|
private String parseXmlAttr(String xml, String attrPrefix) {
|
if (StringUtils.isBlank(xml) || StringUtils.isBlank(attrPrefix)) {
|
return null;
|
}
|
int startIdx = xml.indexOf(attrPrefix);
|
if (startIdx == -1) {
|
return null;
|
}
|
startIdx += attrPrefix.length();
|
int endIdx = xml.indexOf("\"", startIdx);
|
if (endIdx == -1 || startIdx >= endIdx) {
|
return null;
|
}
|
return xml.substring(startIdx, endIdx).trim();
|
}
|
|
// ===================== 工具方法：从XML中解析Sequence_Number为数字 =====================
|
private Integer parseSeqFromXml(String xml) {
|
String seqStr = parseXmlAttr(xml, "Sequence_Number=\"");
|
if (StringUtils.isBlank(seqStr)) {
|
return null;
|
}
|
try {
|
return Integer.parseInt(seqStr);
|
} catch (Exception e) {
|
log.warn("【相机{}】从XML解析Sequence_Number失败 | 值:{}", cameraId, seqStr);
|
return null;
|
}
|
}
|
|
// ===================== 工具方法：查找字节流中的分隔符索引（如\r\n\r\n）=====================
|
private int findDelimiterIndex(ByteBuf buf, byte[] delimiter) {
|
int readerIndex = buf.readerIndex();
|
int readableBytes = buf.readableBytes();
|
int delimiterLen = delimiter.length;
|
|
for (int i = 0; i <= readableBytes - delimiterLen; i++) {
|
boolean match = true;
|
for (int j = 0; j < delimiterLen; j++) {
|
if (buf.getByte(readerIndex + i + j) != delimiter[j]) {
|
match = false;
|
break;
|
}
|
}
|
if (match) {
|
return i;
|
}
|
}
|
return -1;
|
}
|
|
// ===================== 通道事件处理（空闲/断连/异常）=====================
|
@Override
|
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
|
// 30秒读空闲 → 判定设备断连，关闭通道
|
if (evt instanceof IdleStateEvent) {
|
IdleStateEvent event = (IdleStateEvent) evt;
|
if (event.state() == IdleState.READER_IDLE) {
|
log.error("【相机{}】30秒读空闲，判定设备断连，关闭通道", cameraId);
|
ctx.close();
|
// 清理该相机的所有未完成Future，避免内存泄漏
|
ResponseFutureHolder.clearByCamera(cameraId);
|
}
|
}
|
super.userEventTriggered(ctx, evt);
|
}
|
|
// 通道断开 → 清理Future
|
@Override
|
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
log.warn("【相机{}】TCP通道已断开", cameraId);
|
NettyTcpClient.CAMERA_CHANNEL_MAP.remove(cameraId);
|
ResponseFutureHolder.clearByCamera(cameraId);
|
handleTcpConnectFail();
|
super.channelInactive(ctx);
|
}
|
|
/**
|
* 断开处理事件
|
*/
|
private void handleTcpConnectFail() {
|
ArdCamera camera = new ArdCamera();
|
camera.setId(cameraId);
|
camera.setChanNum(0);
|
camera.setLoginId(-1L);
|
camera.setState("0");
|
camera.setChannelList(null);
|
eventPublisher.publishEvent(new LoginEvent(camera));
|
log.warn("设备[{}]TCP连接断开", cameraId);
|
}
|
|
// 通道异常 → 关闭通道+清理Future
|
@Override
|
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
log.error("【相机{}】TCP通道发生异常，关闭通道", cameraId, cause);
|
NettyTcpClient.CAMERA_CHANNEL_MAP.remove(cameraId);
|
ResponseFutureHolder.clearByCamera(cameraId);
|
ctx.close();
|
}
|
|
// 处理器移除 → 释放缓存字节流，避免内存泄漏
|
@Override
|
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
|
if (cacheBuf.isReadable()) {
|
cacheBuf.release();
|
}
|
super.handlerRemoved(ctx);
|
}
|
}
|
