[v0.0.2] 1. InMessageHandler1代码方法抽取；2. TelemetryDecoder空buf释放；3. CacUavParamBindHandler和 CacUavParamQueryHandler向飞控传递数据内容错误修复

pom.xml

@@ -10,7 +10,7 @@
     </parent>
     <groupId>com.platform</groupId>
     <artifactId>pass-through</artifactId>
-    <version>1.0.0</version>
+    <version>1.0.2</version>
     <name>pass-through</name>
     <description>Demo project for Spring Boot</description>
 

src/main/java/com/platform/cac/tcp/message/handler/CacUavParamBindHandler.java

@@ -54,7 +54,7 @@ public class CacUavParamBindHandler implements IRemoteMessageHandler {
             }
             String uavControlUniId = uavCommand.getUavControlUniId();
             log.info("[cac] 中心指控下发参数装订指令...{}", uavCommand);
-            client.writeParamBindCommand(commandCode, commandContent, uavControlUniId);
+            client.writeParamBindCommand(commandCode, command14_29, uavControlUniId);
         } else {
             log.warn("[cac] 当前服务未连接中心指控无法下发参数装订指令");
         }

src/main/java/com/platform/cac/tcp/message/handler/CacUavParamQueryHandler.java

@@ -54,7 +54,7 @@ public class CacUavParamQueryHandler implements IRemoteMessageHandler {
             String uavControlUniId = uavCommand.getUavControlUniId();
             log.info("[cac] 中心指控下发参数查询指令...{}", uavCommand);
 
-            client.writeParamQueryCommand(commandCode, commandContent, uavControlUniId);
+            client.writeParamQueryCommand(commandCode, command14_29, uavControlUniId);
         } else {
             log.warn("[cac] 当前服务未连接中心指控无法下发参数查询指令");
         }

src/main/java/com/platform/service/CacDataRouterHandler.java

@@ -25,7 +25,7 @@ public class CacDataRouterHandler extends SimpleChannelInboundHandler<ByteBuf> {
         if (client == null || client.getClientType() != ClientTypeEnum.HABOR) {
             return;
         }
-        log.debug("[cac] {} 收到数据({}bytes): {}", ((HaborClient) client).shortInfo(), msg.readableBytes(), ByteBufUtil.hexDump(msg));
+        // log.debug("[cac] {} 收到数据({}bytes): {}", ((HaborClient) client).shortInfo(), msg.readableBytes(), ByteBufUtil.hexDump(msg));
         HaborClient haborClient = (HaborClient) client;
         haborClient.process(msg);
     }

src/main/java/com/platform/service/InMessageHandler1.java

@@ -13,7 +13,6 @@ import com.platform.service.clientmanage.ClientManager;
 import com.platform.service.clientmanage.HaborClient;
 import com.platform.util.*;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.*;
 import io.netty.handler.timeout.IdleState;
@@ -120,115 +119,22 @@ public class InMessageHandler1 extends ChannelInboundHandlerAdapter {
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg0) throws Exception {
         String address = getAddress(ctx);
-
         final ByteBuf msgBuf = (ByteBuf) msg0;
-        // 上线握手阶段
-        if (!sessionCache.addToSnMap.containsKey(address)) { // 判断连接是否已经建立，如未建立则先握手
-            try {
-
-                // if (msgBuf.getByte(0) != (byte) 0xAA || msgBuf.getByte(1) != (byte) 0x44) {
-                //     return;
-                // }
-                byte[] msg = new byte[msgBuf.readableBytes()];
-                msgBuf.readBytes(msg);
-                log.info("【接收握手原数据】 " + ByteUtils.bytes2HexString(msg));
-                // 握手数据不满足协议长度，立即返回
-                // String d = stringUtil.bytesToHexString(msg);
-                // byte[] crc = CRC.crc16(msg, 2, 16);
-                // if (crc[0] != msg[18] || crc[1] != msg[19]){
-                //     log.warn("握手数据crc校验错误，期望crc " + stringTool.bytesToHexString(crc) + ", 实际收到 " + d);
-                //     protocolAck(ctx,(byte) 0);
-                //     return;
-                // }
-
-                // 数据类型 1为地面站 2为终端设备
-                String type = new String(msg, 5, 1, Charset.defaultCharset());
-                // deviceCode 终端设备取后六位
-                String deviceCode;
-                String typeName;
-                if (type.equals("1")) {
-                    // 首先读取deviceCode， 哈勃deviceCode是后6个byte
-                    typeName = "地面站";
-                    deviceCode = new String(msg, 6, 12, Charset.defaultCharset()).toUpperCase(); // 地面站
-                    protocolAck(ctx, (byte) 1);
-                } else if (type.equals("2")) {
-                    typeName = "哈勃终端";
-                    deviceCode = new String(msg, 12, 6, Charset.defaultCharset()).toUpperCase(); // 哈勃上线
-                    protocolAck(ctx, (byte) 1);
-                    haborSignIntoCac(deviceCode, ctx.channel());
-                } else {
-                    protocolAck(ctx, (byte) 0); // 握手失败
-                    return;
-                }
-
-                log.info("【读取握手数据】======> deviceCode = " + deviceCode + " type = " + typeName);
-
-                // 删除旧的连接。如果存在异常断开的时候，旧连接并未清除。这里在上线的时候检查并清除
-                String addr0 = sessionCache.snToAddMap.get(deviceCode);
-                if (Objects.nonNull(addr0)) {
-                    sessionCache.tcpSession.remove(addr0);
-                    sessionCache.addToSnMap.remove(addr0);
-                    log.info("【握手删除旧连接】====> " + addr0 + "(" + deviceCode + ")删除");
-                }
-
-                // 添加新连接
-                sessionCache.addToSnMap.put(address, deviceCode);     // ip-mac
-                sessionCache.snToAddMap.put(deviceCode, address);     // mac- ip
-                sessionCache.tcpSession.put(address, ctx.channel());  // ip - channel
 
-                return; // 握手结束立即返回
+        try {
-            } finally {
+            // 上线握手阶段
+            if (!sessionCache.addToSnMap.containsKey(address)) { // 判断连接是否已经建立，如未建立则先握手
+                handshake(ctx, msgBuf, address);
                 msgBuf.release();
+                return; // 握手结束立即返回
             }
-        }
+            else {
-        // 已经建立连接且非地面站控制信息，则进行数据转发
+                // 已经建立连接且非地面站控制信息，则进行数据转发
-        else {
+                transfer(ctx, msgBuf, address);
-            final ByteBuf bufToTransfer = msgBuf.duplicate(); // 转发，保留原始索引
-            String sn = sessionCache.addToSnMap.get(address);    // sn =mac
-            // 监控相关
-            if (!sessionCache.snToMonitorMap.isEmpty()) {
-                Optional<Channel> sendChannelOpt = sessionCache.findMonitorChannel(address);
-                if (sendChannelOpt.isPresent()) {
-                    Channel sendChannel = sendChannelOpt.get();
-                    log.info("(监控)out channel: " + sendChannel.remoteAddress() + "(" + sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/", ""))
-                            + ") 即将发送 --- " + "in channel: " + ctx.channel().remoteAddress() + "(" + sn + ")发来的TCP消息");
-                    sendChannel.writeAndFlush(bufToTransfer.retain()).addListener(
-                            future -> {
-                                // TODO 2025/1/17: 需优化
-                                if (!future.isSuccess()) {
-                                    bufToTransfer.release();
-                                }
-                            }
-                    );
-                }
-
-            }
-
-
-            if (!sessionCache.mappingListMap.isEmpty() && sessionCache.mappingListMap.get(sn) != null) {
-                if (debug) {
-                    // log.info("【转发数据】======> " + ByteBufUtil.hexDump(msgBuf));
-                }
-                List<Channel> sendChannelOpt = sessionCache.findMappingTargetChannel(sn);   // 设备编号对应的通道
-                // log.info("send channel count of {} = {}", sn, sendChannelOpt.size());
-                sendChannelOpt.forEach(sendChannel -> {
-                    // log.info("out channel: " + sendChannel.remoteAddress() + "(" + sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/", "")) + ")即将发送 --- " + "in channel: " + ctx.channel().remoteAddress() + "(" + sn + ")发来的TCP消息");
-
-                    // TODO 2025/1/17: 大量打印日志消耗性能，需要优化，用更合理的方式显示通信对象状态
-                    // log.debug("out channel: {} ({})即将发送 --- from channel: {} ({})", sendChannel.remoteAddress(), sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/", "")), ctx.channel().remoteAddress(), sn);
-                    // 如果需要发送，则增加一次计数引用，便于中心指控继续使用buffer
-                    sendChannel.writeAndFlush(bufToTransfer.retain()).addListener(
-                            future -> {
-                                // TODO 2025/1/17: 需优化
-                                if (!future.isSuccess()) {
-                                    bufToTransfer.release();
-                                } else {
-                                    // log.info("发送成功: {}, {} ", sendChannel.remoteAddress(), ByteBufUtil.hexDump(bufToTransfer));
-                                }
-                            }
-                    );
-                });
             }
+        } catch (RuntimeException e) {
+            ReferenceCountUtil.safeRelease(msgBuf);
+            return;
         }
 
         BaseClient client = ClientManager.getClient(ctx.channel());
@@ -243,6 +149,110 @@ public class InMessageHandler1 extends ChannelInboundHandlerAdapter {
         }
     }
 
+    /**处理哈勃、地面站的握手信息*/
+    private boolean handshake(ChannelHandlerContext ctx, ByteBuf msgBuf, String address) {
+        if (msgBuf.getByte(0) != (byte) 0xAA || msgBuf.getByte(1) != (byte) 0x44) {
+            return false;
+        }
+        byte[] msg = new byte[msgBuf.readableBytes()];
+        msgBuf.readBytes(msg);
+        log.info("【接收握手原数据】 " + ByteUtils.bytes2HexString(msg));
+        // 握手数据不满足协议长度，立即返回
+        // String d = stringUtil.bytesToHexString(msg);
+        // byte[] crc = CRC.crc16(msg, 2, 16);
+        // if (crc[0] != msg[18] || crc[1] != msg[19]){
+        //     log.warn("握手数据crc校验错误，期望crc " + stringTool.bytesToHexString(crc) + ", 实际收到 " + d);
+        //     protocolAck(ctx,(byte) 0);
+        //     return;
+        // }
+
+        // 数据类型 1为地面站 2为终端设备
+        String type = new String(msg, 5, 1, Charset.defaultCharset());
+        // deviceCode 终端设备取后六位
+        String deviceCode;
+        String typeName;
+        if (type.equals("1")) {
+            // 首先读取deviceCode， 哈勃deviceCode是后6个byte
+            typeName = "地面站";
+            deviceCode = new String(msg, 6, 12, Charset.defaultCharset()).toUpperCase(); // 地面站
+            protocolAck(ctx, (byte) 1);
+            log.info("{} {} 握手成功: {}", typeName, deviceCode, ByteUtils.bytes2HexString(msg));
+        } else if (type.equals("2")) {
+            typeName = "哈勃终端";
+            deviceCode = new String(msg, 12, 6, Charset.defaultCharset()).toUpperCase(); // 哈勃上线
+            protocolAck(ctx, (byte) 1);
+            log.info("{} {} 握手成功: {}", typeName, deviceCode, ByteUtils.bytes2HexString(msg));
+            haborSignIntoCac(deviceCode, ctx.channel());
+        } else {
+            protocolAck(ctx, (byte) 0); // 握手失败
+            log.info("握手失败: {}", ByteUtils.bytes2HexString(msg));
+            return false;
+        }
+
+
+        // 删除旧的连接。如果存在异常断开的时候，旧连接并未清除。这里在上线的时候检查并清除
+        String addr0 = sessionCache.snToAddMap.get(deviceCode);
+        if (Objects.nonNull(addr0)) {
+            sessionCache.tcpSession.remove(addr0);
+            sessionCache.addToSnMap.remove(addr0);
+            log.info("【握手删除旧连接】====> " + addr0 + "(" + deviceCode + ")删除");
+        }
+
+        // 添加新连接
+        sessionCache.addToSnMap.put(address, deviceCode);     // ip-mac
+        sessionCache.snToAddMap.put(deviceCode, address);     // mac- ip
+        sessionCache.tcpSession.put(address, ctx.channel());  // ip - channel
+        return true;
+    }
+
+    /**地面站和哈勃之间的数据转发（原哈勃服务功能）*/
+    private void transfer(ChannelHandlerContext ctx, ByteBuf msgBuf, String address) throws RuntimeException {
+        final ByteBuf bufToTransfer = msgBuf.duplicate(); // 转发，保留原始索引
+        String sn = sessionCache.addToSnMap.get(address);    // sn =mac
+
+        // 监控相关
+        if (!sessionCache.snToMonitorMap.isEmpty()) {
+            Optional<Channel> sendChannelOpt = sessionCache.findMonitorChannel(address);
+            if (sendChannelOpt.isPresent()) {
+                Channel sendChannel = sendChannelOpt.get();
+                log.info("(监控)out channel: " + sendChannel.remoteAddress() + "(" + sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/", ""))
+                        + ") 即将发送 --- " + "in channel: " + ctx.channel().remoteAddress() + "(" + sn + ")发来的TCP消息");
+                sendChannel.writeAndFlush(bufToTransfer.retain()).addListener(
+                        future -> {
+                            // TODO 2025/1/17: 需优化
+                            if (!future.isSuccess()) {
+                                bufToTransfer.release();
+                            }
+                        }
+                );
+            }
+
+        }
+
+        if (!sessionCache.mappingListMap.isEmpty() && sessionCache.mappingListMap.get(sn) != null) {
+            if (debug) {
+                // log.info("【转发数据】======> " + ByteBufUtil.hexDump(msgBuf));
+            }
+            List<Channel> sendChannelOpt = sessionCache.findMappingTargetChannel(sn);   // 设备编号对应的通道
+            // log.info("send channel count of {} = {}", sn, sendChannelOpt.size());
+            sendChannelOpt.forEach(sendChannel -> {
+                // log.info("out channel: " + sendChannel.remoteAddress() + "(" + sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/", "")) + ")即将发送 --- " + "in channel: " + ctx.channel().remoteAddress() + "(" + sn + ")发来的TCP消息");
+
+                // TODO 2025/1/17: 大量打印日志消耗性能，需要优化，用更合理的方式显示通信对象状态
+                // log.debug("out channel: {} ({})即将发送 --- from channel: {} ({})", sendChannel.remoteAddress(), sessionCache.addToSnMap.get(sendChannel.remoteAddress().toString().replace("/", "")), ctx.channel().remoteAddress(), sn);
+                // 如果需要发送，则增加一次计数引用，便于中心指控继续使用buffer
+                sendChannel.writeAndFlush(bufToTransfer.retain()).addListener(
+                        future -> {
+                            // TODO 2025/1/17: 需优化
+                            if (!future.isSuccess()) {
+                                bufToTransfer.release();
+                            }
+                        }
+                );
+            });
+        }
+    }
+
     /**
      * 接入中心指控系统, 上线
      */

src/main/java/com/platform/service/TelemetryDecoder.java

@@ -27,10 +27,19 @@ public class TelemetryDecoder extends DelimiterBasedFrameDecoder {
     @Override
     protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
         ByteBuf decodeBuf = (ByteBuf) super.decode(ctx, buffer);
-        if (decodeBuf == null || decodeBuf.readableBytes()==0) {
+        if (decodeBuf == null) {
-            return null; // 如果 decodeBuf 为 null，直接返回 null
+            return null;
+        } else if (decodeBuf.readableBytes()==0){
+            decodeBuf.release(); // 释放空buffer
+            return null;
+        }
+
+        try {
+            // 创建组合buffer，注意HEAD需要复制因为它是静态共享的
+            return Unpooled.wrappedBuffer(HEAD.retainedDuplicate(), decodeBuf);
+        } catch (Exception e) {
+            decodeBuf.release();
+            return null; // Add return statement in catch block
         }
-        return Unpooled.wrappedBuffer(HEAD.retain(), decodeBuf); // 只有在 decodeBuf 不为 null 时，才补上帧头
-        // return decodeBuf;
     }
 }

src/main/java/com/platform/service/clientmanage/HaborClient981A.java

@@ -241,6 +241,11 @@ public class HaborClient981A extends HaborClient {
     @Override
     public void writeParamBindCommand(byte commandCode, byte[] commandContent, String controlUniId) {
         // try {
+        // byte[] test = {
+        //         (byte) 0x39, (byte) 0x32, (byte) 0x00, (byte) 0x36, (byte) 0x32, (byte) 0x14, (byte) 0xF6, (byte) 0x00, (byte) 0x00, (byte) 0x28, (byte) 0x05, (byte) 0x05, (byte) 0x00, (byte) 0x00, (byte) 0x04, (byte) 0x00
+        // };
+        // 39 32 00 36 32 14 F6 00 00 28 05 05 00 00 04 00
+        // 39 32 00 36 32 14 F6 00 00 28 05 05 00 05 00 03
         bindDataBuf.buf.setBytes(13, commandContent);
         setCurrentSendData(bindDataBuf); // 设置当前发送的数据类型
         previousParamBind.recordCommand(commandCode, controlUniId, bindDataBuf.buf.slice(13,16)); // 记录参数装订指令

src/main/resources/application.yml

@@ -1,6 +1,6 @@
 spring:
   profiles:
-    active: pre
+    active: dev
 
 app:
   debug: true