Commit 5ead81beb1d6560815f86b680486fa9266ce95cb

Authored by YunaiV
1 parent 046b2d51

feat(iot):【协议改造】tcp 初步改造(90%):优化 codec 代码

Showing 13 changed files with 156 additions and 169 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotTcpConfig.java
@@ -64,13 +64,13 @@ public class IotTcpConfig { @@ -64,13 +64,13 @@ public class IotTcpConfig {
64 /** 64 /**
65 * LENGTH_FIELD: 长度字段偏移量 65 * LENGTH_FIELD: 长度字段偏移量
66 * <p> 66 * <p>
67 - * 表示长度字段在消息中的起始位置(从0开始) 67 + * 表示长度字段在消息中的起始位置(从 0 开始)
68 */ 68 */
69 private Integer lengthFieldOffset; 69 private Integer lengthFieldOffset;
70 /** 70 /**
71 * LENGTH_FIELD: 长度字段长度(字节数) 71 * LENGTH_FIELD: 长度字段长度(字节数)
72 * <p> 72 * <p>
73 - * 常见值:1(最大255)、2(最大65535)、4(最大2GB) 73 + * 常见值:1(最大 255)、2(最大 65535)、4(最大 2GB)
74 */ 74 */
75 private Integer lengthFieldLength; 75 private Integer lengthFieldLength;
76 /** 76 /**
@@ -100,15 +100,6 @@ public class IotTcpConfig { @@ -100,15 +100,6 @@ public class IotTcpConfig {
100 */ 100 */
101 private Integer fixedLength; 101 private Integer fixedLength;
102 102
103 - /**  
104 - * 最大帧长度(字节)  
105 - * <p>  
106 - * 防止内存溢出,默认 1MB  
107 - */  
108 - @NotNull(message = "最大帧长度不能为空")  
109 - @Min(value = 1, message = "最大帧长度必须大于 0")  
110 - private Integer maxFrameLength = 1048576;  
111 -  
112 } 103 }
113 104
114 } 105 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotTcpProtocol.java
@@ -7,6 +7,7 @@ import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; @@ -7,6 +7,7 @@ import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
7 import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties.ProtocolInstanceProperties; 7 import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties.ProtocolInstanceProperties;
8 import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol; 8 import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol;
9 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec; 9 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec;
  10 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodecFactory;
10 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.downstream.IotTcpDownstreamHandler; 11 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.downstream.IotTcpDownstreamHandler;
11 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.downstream.IotTcpDownstreamSubscriber; 12 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.downstream.IotTcpDownstreamSubscriber;
12 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.upstream.IotTcpUpstreamHandler; 13 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.upstream.IotTcpUpstreamHandler;
@@ -89,7 +90,7 @@ public class IotTcpProtocol implements IotProtocol { @@ -89,7 +90,7 @@ public class IotTcpProtocol implements IotProtocol {
89 // 初始化帧编解码器 90 // 初始化帧编解码器
90 IotTcpConfig tcpConfig = properties.getTcp(); 91 IotTcpConfig tcpConfig = properties.getTcp();
91 IotTcpConfig.CodecConfig codecConfig = tcpConfig != null ? tcpConfig.getCodec() : null; 92 IotTcpConfig.CodecConfig codecConfig = tcpConfig != null ? tcpConfig.getCodec() : null;
92 - this.frameCodec = IotTcpFrameCodec.create(codecConfig); 93 + this.frameCodec = IotTcpFrameCodecFactory.create(codecConfig);
93 94
94 // 初始化连接管理器 95 // 初始化连接管理器
95 this.connectionManager = new IotTcpConnectionManager(); 96 this.connectionManager = new IotTcpConnectionManager();
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/IotTcpCodecTypeEnum.java
1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec; 1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec;
2 2
3 import cn.hutool.core.util.ArrayUtil; 3 import cn.hutool.core.util.ArrayUtil;
4 -import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpConfig;  
5 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.delimiter.IotTcpDelimiterFrameCodec; 4 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.delimiter.IotTcpDelimiterFrameCodec;
6 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.length.IotTcpFixedLengthFrameCodec; 5 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.length.IotTcpFixedLengthFrameCodec;
7 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.length.IotTcpLengthFieldFrameCodec; 6 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.length.IotTcpLengthFieldFrameCodec;
8 import lombok.AllArgsConstructor; 7 import lombok.AllArgsConstructor;
9 import lombok.Getter; 8 import lombok.Getter;
10 9
11 -import java.util.function.Function;  
12 -  
13 /** 10 /**
14 * IoT TCP 拆包类型枚举 11 * IoT TCP 拆包类型枚举
15 * 12 *
@@ -21,38 +18,28 @@ public enum IotTcpCodecTypeEnum { @@ -21,38 +18,28 @@ public enum IotTcpCodecTypeEnum {
21 18
22 /** 19 /**
23 * 基于固定长度的拆包 20 * 基于固定长度的拆包
24 - * <p>  
25 - * 消息格式:固定长度的消息体  
26 - * 需要配置:fixedLength(固定长度)  
27 */ 21 */
28 - FIXED_LENGTH("fixed_length", IotTcpFixedLengthFrameCodec::new), 22 + FIXED_LENGTH("fixed_length", IotTcpFixedLengthFrameCodec.class),
29 23
30 /** 24 /**
31 * 基于分隔符的拆包 25 * 基于分隔符的拆包
32 - * <p>  
33 - * 消息格式:消息内容 + 分隔符  
34 - * 需要配置:delimiter(分隔符)  
35 */ 26 */
36 - DELIMITER("delimiter", IotTcpDelimiterFrameCodec::new), 27 + DELIMITER("delimiter", IotTcpDelimiterFrameCodec.class),
37 28
38 /** 29 /**
39 * 基于长度字段的拆包 30 * 基于长度字段的拆包
40 - * <p>  
41 - * 消息格式:[长度字段][消息体]  
42 - * 需要配置:lengthFieldOffset(长度字段偏移量)、lengthFieldLength(长度字段长度)  
43 */ 31 */
44 - LENGTH_FIELD("length_field", IotTcpLengthFieldFrameCodec::new), 32 + LENGTH_FIELD("length_field", IotTcpLengthFieldFrameCodec.class),
45 ; 33 ;
46 34
47 /** 35 /**
48 * 类型标识 36 * 类型标识
49 */ 37 */
50 private final String type; 38 private final String type;
51 -  
52 /** 39 /**
53 - * Codec 创建工厂 40 + * 编解码器类
54 */ 41 */
55 - private final Function<IotTcpConfig.CodecConfig, IotTcpFrameCodec> codecFactory; 42 + private final Class<? extends IotTcpFrameCodec> codecClass;
56 43
57 /** 44 /**
58 * 根据类型获取枚举 45 * 根据类型获取枚举
@@ -64,14 +51,4 @@ public enum IotTcpCodecTypeEnum { @@ -64,14 +51,4 @@ public enum IotTcpCodecTypeEnum {
64 return ArrayUtil.firstMatch(e -> e.getType().equalsIgnoreCase(type), values()); 51 return ArrayUtil.firstMatch(e -> e.getType().equalsIgnoreCase(type), values());
65 } 52 }
66 53
67 - /**  
68 - * 创建 Codec 实例  
69 - *  
70 - * @param config 拆包配置  
71 - * @return Codec 实例  
72 - */  
73 - public IotTcpFrameCodec createCodec(IotTcpConfig.CodecConfig config) {  
74 - return codecFactory.apply(config);  
75 - }  
76 -  
77 } 54 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/IotTcpFrameCodec.java
1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec; 1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec;
2 2
3 -import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpConfig;  
4 import io.vertx.core.Handler; 3 import io.vertx.core.Handler;
5 import io.vertx.core.buffer.Buffer; 4 import io.vertx.core.buffer.Buffer;
6 import io.vertx.core.parsetools.RecordParser; 5 import io.vertx.core.parsetools.RecordParser;
@@ -41,24 +40,4 @@ public interface IotTcpFrameCodec { @@ -41,24 +40,4 @@ public interface IotTcpFrameCodec {
41 */ 40 */
42 Buffer encode(byte[] data); 41 Buffer encode(byte[] data);
43 42
44 - // TODO @AI:还是搞个 facory 类 ,更好理解;  
45 - // ========== 静态工厂方法 ==========  
46 -  
47 - /**  
48 - * 根据配置创建编解码器  
49 - *  
50 - * @param config 拆包配置  
51 - * @return 编解码器实例,如果配置为空则返回 null  
52 - */  
53 - static IotTcpFrameCodec create(IotTcpConfig.CodecConfig config) {  
54 - if (config == null) {  
55 - return null;  
56 - }  
57 - IotTcpCodecTypeEnum type = IotTcpCodecTypeEnum.of(config.getType());  
58 - if (type == null) {  
59 - return null;  
60 - }  
61 - return type.createCodec(config);  
62 - }  
63 -  
64 } 43 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/IotTcpFrameCodecFactory.java 0 → 100644
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec;
  2 +
  3 +import cn.hutool.core.lang.Assert;
  4 +import cn.hutool.core.util.ReflectUtil;
  5 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpConfig;
  6 +
  7 +/**
  8 + * IoT TCP 帧编解码器工厂
  9 + *
  10 + * @author 芋道源码
  11 + */
  12 +public class IotTcpFrameCodecFactory {
  13 +
  14 + /**
  15 + * 根据配置创建编解码器
  16 + *
  17 + * @param config 拆包配置
  18 + * @return 编解码器实例,如果配置为空则返回 null
  19 + */
  20 + public static IotTcpFrameCodec create(IotTcpConfig.CodecConfig config) {
  21 + Assert.notNull(config, "CodecConfig 不能为空");
  22 + IotTcpCodecTypeEnum type = IotTcpCodecTypeEnum.of(config.getType());
  23 + Assert.notNull(type, "不支持的 CodecType 类型:" + config.getType());
  24 + return ReflectUtil.newInstance(type.getCodecClass(), config);
  25 + }
  26 +
  27 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/delimiter/IotTcpDelimiterFrameCodec.java
1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.delimiter; 1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.delimiter;
2 2
3 -import cn.hutool.core.util.StrUtil;  
4 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpConfig; 3 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpConfig;
5 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpCodecTypeEnum; 4 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpCodecTypeEnum;
6 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec; 5 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec;
@@ -8,6 +7,7 @@ import io.vertx.core.Handler; @@ -8,6 +7,7 @@ import io.vertx.core.Handler;
8 import io.vertx.core.buffer.Buffer; 7 import io.vertx.core.buffer.Buffer;
9 import io.vertx.core.parsetools.RecordParser; 8 import io.vertx.core.parsetools.RecordParser;
10 import lombok.extern.slf4j.Slf4j; 9 import lombok.extern.slf4j.Slf4j;
  10 +import org.springframework.util.Assert;
11 11
12 /** 12 /**
13 * IoT TCP 分隔符帧编解码器 13 * IoT TCP 分隔符帧编解码器
@@ -27,24 +27,14 @@ import lombok.extern.slf4j.Slf4j; @@ -27,24 +27,14 @@ import lombok.extern.slf4j.Slf4j;
27 @Slf4j 27 @Slf4j
28 public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec { 28 public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec {
29 29
30 - private final IotTcpConfig.CodecConfig config;  
31 -  
32 /** 30 /**
33 * 解析后的分隔符字节数组 31 * 解析后的分隔符字节数组
34 */ 32 */
35 private final byte[] delimiterBytes; 33 private final byte[] delimiterBytes;
36 34
37 - /**  
38 - * 最大帧长度  
39 - */  
40 - // TODO @AI:最大帧数要不去掉;简洁一点;包括其他地方的配置项;  
41 - private final int maxFrameLength;  
42 -  
43 public IotTcpDelimiterFrameCodec(IotTcpConfig.CodecConfig config) { 35 public IotTcpDelimiterFrameCodec(IotTcpConfig.CodecConfig config) {
44 - this.config = config;  
45 - // TODO @AI:禁止为空; 36 + Assert.hasText(config.getDelimiter(), "delimiter 不能为空");
46 this.delimiterBytes = parseDelimiter(config.getDelimiter()); 37 this.delimiterBytes = parseDelimiter(config.getDelimiter());
47 - this.maxFrameLength = config.getMaxFrameLength() != null ? config.getMaxFrameLength() : 1048576;  
48 } 38 }
49 39
50 @Override 40 @Override
@@ -55,20 +45,11 @@ public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec { @@ -55,20 +45,11 @@ public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec {
55 @Override 45 @Override
56 public RecordParser createDecodeParser(Handler<Buffer> handler) { 46 public RecordParser createDecodeParser(Handler<Buffer> handler) {
57 RecordParser parser = RecordParser.newDelimited(Buffer.buffer(delimiterBytes)); 47 RecordParser parser = RecordParser.newDelimited(Buffer.buffer(delimiterBytes));
58 -  
59 - parser.handler(buffer -> {  
60 - // 检查帧长度是否超过限制  
61 - if (buffer.length() > maxFrameLength) {  
62 - log.warn("[createDecodeParser][帧长度超过限制,length: {}, maxFrameLength: {}]",  
63 - buffer.length(), maxFrameLength);  
64 - return;  
65 - }  
66 - // 处理完整消息(不包含分隔符)  
67 - handler.handle(buffer); 48 + // 处理完整消息(不包含分隔符)
  49 + parser.handler(handler);
  50 + parser.exceptionHandler(ex -> {
  51 + throw new RuntimeException("[createDecodeParser][解析异常]", ex);
68 }); 52 });
69 -  
70 - // TODO @AI:异常处理;  
71 - parser.exceptionHandler(ex -> log.error("[createDecodeParser][解析异常]", ex));  
72 return parser; 53 return parser;
73 } 54 }
74 55
@@ -89,13 +70,7 @@ public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec { @@ -89,13 +70,7 @@ public class IotTcpDelimiterFrameCodec implements IotTcpFrameCodec {
89 * @return 分隔符字节数组 70 * @return 分隔符字节数组
90 */ 71 */
91 private byte[] parseDelimiter(String delimiter) { 72 private byte[] parseDelimiter(String delimiter) {
92 - if (StrUtil.isBlank(delimiter)) {  
93 - // 默认使用换行符  
94 - return new byte[]{'\n'};  
95 - }  
96 -  
97 // 处理转义字符 73 // 处理转义字符
98 - // TODO @AI:是否必要?不调整感觉也没问题?用户自己写对就 ok 了是哇?  
99 String parsed = delimiter 74 String parsed = delimiter
100 .replace("\\r\\n", "\r\n") 75 .replace("\\r\\n", "\r\n")
101 .replace("\\r", "\r") 76 .replace("\\r", "\r")
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/length/IotTcpFixedLengthFrameCodec.java
@@ -7,6 +7,7 @@ import io.vertx.core.Handler; @@ -7,6 +7,7 @@ import io.vertx.core.Handler;
7 import io.vertx.core.buffer.Buffer; 7 import io.vertx.core.buffer.Buffer;
8 import io.vertx.core.parsetools.RecordParser; 8 import io.vertx.core.parsetools.RecordParser;
9 import lombok.extern.slf4j.Slf4j; 9 import lombok.extern.slf4j.Slf4j;
  10 +import org.springframework.util.Assert;
10 11
11 /** 12 /**
12 * IoT TCP 定长帧编解码器 13 * IoT TCP 定长帧编解码器
@@ -24,8 +25,8 @@ public class IotTcpFixedLengthFrameCodec implements IotTcpFrameCodec { @@ -24,8 +25,8 @@ public class IotTcpFixedLengthFrameCodec implements IotTcpFrameCodec {
24 private final int fixedLength; 25 private final int fixedLength;
25 26
26 public IotTcpFixedLengthFrameCodec(IotTcpConfig.CodecConfig config) { 27 public IotTcpFixedLengthFrameCodec(IotTcpConfig.CodecConfig config) {
27 - // TODO @AI:config.getFixedLength() 禁止为空;  
28 - this.fixedLength = config.getFixedLength() != null ? config.getFixedLength() : 1024; 28 + Assert.notNull(config.getFixedLength(), "fixedLength 不能为空");
  29 + this.fixedLength = config.getFixedLength();
29 } 30 }
30 31
31 @Override 32 @Override
@@ -37,8 +38,9 @@ public class IotTcpFixedLengthFrameCodec implements IotTcpFrameCodec { @@ -37,8 +38,9 @@ public class IotTcpFixedLengthFrameCodec implements IotTcpFrameCodec {
37 public RecordParser createDecodeParser(Handler<Buffer> handler) { 38 public RecordParser createDecodeParser(Handler<Buffer> handler) {
38 RecordParser parser = RecordParser.newFixed(fixedLength); 39 RecordParser parser = RecordParser.newFixed(fixedLength);
39 parser.handler(handler); 40 parser.handler(handler);
40 - // TODO @AI:解析失败,是不是要抛出异常?因为要 close 掉连接;  
41 - parser.exceptionHandler(ex -> log.error("[createDecodeParser][解析异常]", ex)); 41 + parser.exceptionHandler(ex -> {
  42 + throw new RuntimeException("[createDecodeParser][解析异常]", ex);
  43 + });
42 return parser; 44 return parser;
43 } 45 }
44 46
@@ -46,8 +48,7 @@ public class IotTcpFixedLengthFrameCodec implements IotTcpFrameCodec { @@ -46,8 +48,7 @@ public class IotTcpFixedLengthFrameCodec implements IotTcpFrameCodec {
46 public Buffer encode(byte[] data) { 48 public Buffer encode(byte[] data) {
47 Buffer buffer = Buffer.buffer(fixedLength); 49 Buffer buffer = Buffer.buffer(fixedLength);
48 buffer.appendBytes(data); 50 buffer.appendBytes(data);
49 - // 如果数据不足固定长度,填充 0  
50 - // TODO @AI:这里的填充是合理的么?RecordParser.newFixed(fixedLength) 有填充的逻辑么? 51 + // 如果数据不足固定长度,填充 0(RecordParser.newFixed 解码时按固定长度读取,所以发送端需要填充)
51 if (data.length < fixedLength) { 52 if (data.length < fixedLength) {
52 byte[] padding = new byte[fixedLength - data.length]; 53 byte[] padding = new byte[fixedLength - data.length];
53 buffer.appendBytes(padding); 54 buffer.appendBytes(padding);
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/codec/length/IotTcpLengthFieldFrameCodec.java
@@ -7,6 +7,9 @@ import io.vertx.core.Handler; @@ -7,6 +7,9 @@ import io.vertx.core.Handler;
7 import io.vertx.core.buffer.Buffer; 7 import io.vertx.core.buffer.Buffer;
8 import io.vertx.core.parsetools.RecordParser; 8 import io.vertx.core.parsetools.RecordParser;
9 import lombok.extern.slf4j.Slf4j; 9 import lombok.extern.slf4j.Slf4j;
  10 +import org.springframework.util.Assert;
  11 +
  12 +import java.util.concurrent.atomic.AtomicReference;
10 13
11 /** 14 /**
12 * IoT TCP 长度字段帧编解码器 15 * IoT TCP 长度字段帧编解码器
@@ -30,8 +33,6 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec { @@ -30,8 +33,6 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec {
30 private final int lengthFieldLength; 33 private final int lengthFieldLength;
31 private final int lengthAdjustment; 34 private final int lengthAdjustment;
32 private final int initialBytesToStrip; 35 private final int initialBytesToStrip;
33 - // TODO @AI:去掉 maxFrameLength 相关字段;  
34 - private final int maxFrameLength;  
35 36
36 /** 37 /**
37 * 头部长度 = 长度字段偏移量 + 长度字段长度 38 * 头部长度 = 长度字段偏移量 + 长度字段长度
@@ -39,12 +40,14 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec { @@ -39,12 +40,14 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec {
39 private final int headerLength; 40 private final int headerLength;
40 41
41 public IotTcpLengthFieldFrameCodec(IotTcpConfig.CodecConfig config) { 42 public IotTcpLengthFieldFrameCodec(IotTcpConfig.CodecConfig config) {
42 - // TODO @AI: 增加参数校验;不要 default 逻辑;  
43 - this.lengthFieldOffset = config.getLengthFieldOffset() != null ? config.getLengthFieldOffset() : 0;  
44 - this.lengthFieldLength = config.getLengthFieldLength() != null ? config.getLengthFieldLength() : 4;  
45 - this.lengthAdjustment = config.getLengthAdjustment() != null ? config.getLengthAdjustment() : 0;  
46 - this.initialBytesToStrip = config.getInitialBytesToStrip() != null ? config.getInitialBytesToStrip() : 0;  
47 - this.maxFrameLength = config.getMaxFrameLength() != null ? config.getMaxFrameLength() : 1048576; 43 + Assert.notNull(config.getLengthFieldOffset(), "lengthFieldOffset 不能为空");
  44 + Assert.notNull(config.getLengthFieldLength(), "lengthFieldLength 不能为空");
  45 + Assert.notNull(config.getLengthAdjustment(), "lengthAdjustment 不能为空");
  46 + Assert.notNull(config.getInitialBytesToStrip(), "initialBytesToStrip 不能为空");
  47 + this.lengthFieldOffset = config.getLengthFieldOffset();
  48 + this.lengthFieldLength = config.getLengthFieldLength();
  49 + this.lengthAdjustment = config.getLengthAdjustment();
  50 + this.initialBytesToStrip = config.getInitialBytesToStrip();
48 this.headerLength = lengthFieldOffset + lengthFieldLength; 51 this.headerLength = lengthFieldOffset + lengthFieldLength;
49 } 52 }
50 53
@@ -57,49 +60,45 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec { @@ -57,49 +60,45 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec {
57 public RecordParser createDecodeParser(Handler<Buffer> handler) { 60 public RecordParser createDecodeParser(Handler<Buffer> handler) {
58 // 创建状态机:先读取头部,再读取消息体 61 // 创建状态机:先读取头部,再读取消息体
59 RecordParser parser = RecordParser.newFixed(headerLength); 62 RecordParser parser = RecordParser.newFixed(headerLength);
60 - // 使用数组保存状态和头部数据  
61 - // TODO @AI:bodyLength 只使用第 0 位,是不是 atomicInteger 更合适?  
62 - final int[] bodyLength = {-1};  
63 - final Buffer[] headerBuffer = {null}; 63 + final AtomicReference<Integer> bodyLength = new AtomicReference<>(null); // 消息体长度,null 表示读取头部阶段
  64 + final AtomicReference<Buffer> headerBuffer = new AtomicReference<>(null); // 头部消息
64 65
65 // 处理读取到的数据 66 // 处理读取到的数据
66 parser.handler(buffer -> { 67 parser.handler(buffer -> {
67 - if (bodyLength[0] == -1) { 68 + if (bodyLength.get() == null) {
68 // 阶段 1: 读取头部,解析长度字段 69 // 阶段 1: 读取头部,解析长度字段
69 - headerBuffer[0] = buffer.copy(); 70 + headerBuffer.set(buffer.copy());
70 int length = readLength(buffer, lengthFieldOffset, lengthFieldLength); 71 int length = readLength(buffer, lengthFieldOffset, lengthFieldLength);
71 int frameBodyLength = length + lengthAdjustment; 72 int frameBodyLength = length + lengthAdjustment;
72 - // 检查帧长度是否超过限制  
73 - if (frameBodyLength < 0 || frameBodyLength > maxFrameLength - headerLength) {  
74 - log.warn("[createDecodeParser][帧长度异常,length: {}, frameBodyLength: {}, maxFrameLength: {}]",  
75 - length, frameBodyLength, maxFrameLength);  
76 - return; 73 + // 检查帧长度是否合法
  74 + if (frameBodyLength < 0) {
  75 + throw new IllegalStateException(String.format(
  76 + "[createDecodeParser][帧长度异常,length: %d, frameBodyLength: %d]",
  77 + length, frameBodyLength));
77 } 78 }
78 - 79 + // 消息体为空,抛出异常
79 if (frameBodyLength == 0) { 80 if (frameBodyLength == 0) {
80 - // 消息体为空,直接处理  
81 - // TODO @AI:消息体为空,是不是不合理哈?应该抛出异常?  
82 - Buffer frame = processFrame(headerBuffer[0], null);  
83 - handler.handle(frame);  
84 - } else {  
85 - // 切换到读取消息体模式  
86 - bodyLength[0] = frameBodyLength;  
87 - parser.fixedSizeMode(frameBodyLength); 81 + throw new IllegalStateException("[createDecodeParser][消息体不能为空]");
88 } 82 }
  83 +
  84 + // 【重要】切换到读取消息体模式
  85 + bodyLength.set(frameBodyLength);
  86 + parser.fixedSizeMode(frameBodyLength);
89 } else { 87 } else {
90 // 阶段 2: 读取消息体,组装完整帧 88 // 阶段 2: 读取消息体,组装完整帧
91 - Buffer frame = processFrame(headerBuffer[0], buffer); 89 + Buffer frame = processFrame(headerBuffer.get(), buffer);
92 // 重置状态,准备读取下一帧 90 // 重置状态,准备读取下一帧
93 - bodyLength[0] = -1;  
94 - headerBuffer[0] = null; 91 + bodyLength.set(null);
  92 + headerBuffer.set(null);
95 parser.fixedSizeMode(headerLength); 93 parser.fixedSizeMode(headerLength);
96 94
97 - // 处理完整消息 95 + // 【重要】处理完整消息
98 handler.handle(frame); 96 handler.handle(frame);
99 } 97 }
100 }); 98 });
101 -  
102 - parser.exceptionHandler(ex -> log.error("[createDecodeParser][解析异常]", ex)); 99 + parser.exceptionHandler(ex -> {
  100 + throw new RuntimeException("[createDecodeParser][解析异常]", ex);
  101 + });
103 return parser; 102 return parser;
104 } 103 }
105 104
@@ -122,26 +121,36 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec { @@ -122,26 +121,36 @@ public class IotTcpLengthFieldFrameCodec implements IotTcpFrameCodec {
122 /** 121 /**
123 * 从 Buffer 中读取长度字段 122 * 从 Buffer 中读取长度字段
124 */ 123 */
125 - // TODO @AI:兼容 JDK8 124 + @SuppressWarnings("EnhancedSwitchMigration")
126 private int readLength(Buffer buffer, int offset, int length) { 125 private int readLength(Buffer buffer, int offset, int length) {
127 - return switch (length) {  
128 - case 1 -> buffer.getUnsignedByte(offset);  
129 - case 2 -> buffer.getUnsignedShort(offset);  
130 - case 4 -> buffer.getInt(offset);  
131 - default -> throw new IllegalArgumentException("不支持的长度字段长度: " + length);  
132 - }; 126 + switch (length) {
  127 + case 1:
  128 + return buffer.getUnsignedByte(offset);
  129 + case 2:
  130 + return buffer.getUnsignedShort(offset);
  131 + case 4:
  132 + return buffer.getInt(offset);
  133 + default:
  134 + throw new IllegalArgumentException("不支持的长度字段长度: " + length);
  135 + }
133 } 136 }
134 137
135 /** 138 /**
136 * 向 Buffer 中写入长度字段 139 * 向 Buffer 中写入长度字段
137 */ 140 */
138 - // TODO @AI:兼容 JDK8  
139 private void writeLength(Buffer buffer, int length, int fieldLength) { 141 private void writeLength(Buffer buffer, int length, int fieldLength) {
140 switch (fieldLength) { 142 switch (fieldLength) {
141 - case 1 -> buffer.appendByte((byte) length);  
142 - case 2 -> buffer.appendShort((short) length);  
143 - case 4 -> buffer.appendInt(length);  
144 - default -> throw new IllegalArgumentException("不支持的长度字段长度: " + fieldLength); 143 + case 1:
  144 + buffer.appendByte((byte) length);
  145 + break;
  146 + case 2:
  147 + buffer.appendShort((short) length);
  148 + break;
  149 + case 4:
  150 + buffer.appendInt(length);
  151 + break;
  152 + default:
  153 + throw new IllegalArgumentException("不支持的长度字段长度: " + fieldLength);
145 } 154 }
146 } 155 }
147 156
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/downstream/IotTcpDownstreamHandler.java
@@ -39,8 +39,8 @@ public class IotTcpDownstreamHandler { @@ -39,8 +39,8 @@ public class IotTcpDownstreamHandler {
39 IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfoByDeviceId( 39 IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfoByDeviceId(
40 message.getDeviceId()); 40 message.getDeviceId());
41 if (connectionInfo == null) { 41 if (connectionInfo == null) {
42 - // TODO @AI:是不是把消息 id 也打印进去?类似上面的日志  
43 - log.warn("[handle][连接信息不存在,设备 ID: {}]", message.getDeviceId()); 42 + log.warn("[handle][连接信息不存在,设备 ID: {},方法: {},消息 ID: {}]",
  43 + message.getDeviceId(), message.getMethod(), message.getId());
44 return; 44 return;
45 } 45 }
46 46
@@ -50,14 +50,11 @@ public class IotTcpDownstreamHandler { @@ -50,14 +50,11 @@ public class IotTcpDownstreamHandler {
50 50
51 // 3. 发送到设备 51 // 3. 发送到设备
52 boolean success = connectionManager.sendToDevice(message.getDeviceId(), frameData.getBytes()); 52 boolean success = connectionManager.sendToDevice(message.getDeviceId(), frameData.getBytes());
53 - // TODO @AI:不成功,直接抛出异常;反正下面的日志也会打印失败的  
54 - if (success) {  
55 - log.info("[handle][下行消息发送成功,设备 ID: {},方法: {},消息 ID: {},数据长度: {} 字节]",  
56 - message.getDeviceId(), message.getMethod(), message.getId(), frameData.length());  
57 - } else {  
58 - log.error("[handle][下行消息发送失败,设备 ID: {},方法: {},消息 ID: {}]",  
59 - message.getDeviceId(), message.getMethod(), message.getId()); 53 + if (!success) {
  54 + throw new RuntimeException("下行消息发送失败");
60 } 55 }
  56 + log.info("[handle][下行消息发送成功,设备 ID: {},方法: {},消息 ID: {},数据长度: {} 字节]",
  57 + message.getDeviceId(), message.getMethod(), message.getId(), frameData.length());
61 } catch (Exception e) { 58 } catch (Exception e) {
62 log.error("[handle][处理下行消息失败,设备 ID: {},方法: {},消息内容: {}]", 59 log.error("[handle][处理下行消息失败,设备 ID: {},方法: {},消息内容: {}]",
63 message.getDeviceId(), message.getMethod(), message, e); 60 message.getDeviceId(), message.getMethod(), message, e);
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
@@ -68,7 +68,13 @@ yudao: @@ -68,7 +68,13 @@ yudao:
68 codec: 68 codec:
69 type: delimiter # 拆包类型:length_field / delimiter / fixed_length 69 type: delimiter # 拆包类型:length_field / delimiter / fixed_length
70 delimiter: "\\n" # 分隔符(支持转义:\\n=换行, \\r=回车, \\t=制表符) 70 delimiter: "\\n" # 分隔符(支持转义:\\n=换行, \\r=回车, \\t=制表符)
71 - max-frame-length: 1048576 # 最大帧长度(字节) 71 +# type: length_field # 拆包类型:length_field / delimiter / fixed_length
  72 +# length-field-offset: 0 # 长度字段偏移量
  73 +# length-field-length: 4 # 长度字段长度
  74 +# length-adjustment: 0 # 长度调整值
  75 +# initial-bytes-to-strip: 4 # 初始跳过的字节数
  76 +# type: fixed_length # 拆包类型:length_field / delimiter / fixed_length
  77 +# fixed-length: 256 # 固定长度
72 78
73 # 协议配置(旧版,保持兼容) 79 # 协议配置(旧版,保持兼容)
74 protocol: 80 protocol:
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotDirectDeviceTcpProtocolIntegrationTest.java
@@ -8,7 +8,9 @@ import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO; @@ -8,7 +8,9 @@ import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO;
8 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO; 8 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO;
9 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO; 9 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO;
10 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; 10 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  11 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpCodecTypeEnum;
11 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec; 12 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec;
  13 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodecFactory;
12 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer; 14 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
13 import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer; 15 import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer;
14 import io.vertx.core.Vertx; 16 import io.vertx.core.Vertx;
@@ -69,11 +71,17 @@ public class IotDirectDeviceTcpProtocolIntegrationTest { @@ -69,11 +71,17 @@ public class IotDirectDeviceTcpProtocolIntegrationTest {
69 /** 71 /**
70 * TCP 帧编解码器 72 * TCP 帧编解码器
71 */ 73 */
72 - private static final IotTcpFrameCodec FRAME_CODEC = IotTcpFrameCodec.create(  
73 - new IotTcpConfig.CodecConfig() {{  
74 - setType("delimiter");  
75 - setDelimiter("\\n");  
76 - }} 74 + private static final IotTcpFrameCodec FRAME_CODEC = IotTcpFrameCodecFactory.create(
  75 + new IotTcpConfig.CodecConfig()
  76 + .setType(IotTcpCodecTypeEnum.DELIMITER.getType())
  77 + .setDelimiter("\\n")
  78 +// .setType(IotTcpCodecTypeEnum.LENGTH_FIELD.getType())
  79 +// .setLengthFieldOffset(0)
  80 +// .setLengthFieldLength(4)
  81 +// .setLengthAdjustment(0)
  82 +// .setInitialBytesToStrip(4)
  83 +// .setType(IotTcpCodecTypeEnum.LENGTH_FIELD.getType())
  84 +// .setFixedLength(256)
77 ); 85 );
78 86
79 // ===================== 直连设备信息(根据实际情况修改,从 iot_device 表查询) ===================== 87 // ===================== 直连设备信息(根据实际情况修改,从 iot_device 表查询) =====================
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotGatewayDeviceTcpProtocolIntegrationTest.java
@@ -13,7 +13,9 @@ import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoAddReqDTO; @@ -13,7 +13,9 @@ import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoAddReqDTO;
13 import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoDeleteReqDTO; 13 import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoDeleteReqDTO;
14 import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoGetReqDTO; 14 import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoGetReqDTO;
15 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; 15 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  16 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpCodecTypeEnum;
16 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec; 17 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec;
  18 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodecFactory;
17 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer; 19 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
18 import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer; 20 import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer;
19 import io.vertx.core.Vertx; 21 import io.vertx.core.Vertx;
@@ -78,11 +80,17 @@ public class IotGatewayDeviceTcpProtocolIntegrationTest { @@ -78,11 +80,17 @@ public class IotGatewayDeviceTcpProtocolIntegrationTest {
78 /** 80 /**
79 * TCP 帧编解码器 81 * TCP 帧编解码器
80 */ 82 */
81 - private static final IotTcpFrameCodec FRAME_CODEC = IotTcpFrameCodec.create(  
82 - new IotTcpConfig.CodecConfig() {{  
83 - setType("delimiter");  
84 - setDelimiter("\\n");  
85 - }} 83 + private static final IotTcpFrameCodec FRAME_CODEC = IotTcpFrameCodecFactory.create(
  84 + new IotTcpConfig.CodecConfig()
  85 + .setType(IotTcpCodecTypeEnum.DELIMITER.getType())
  86 + .setDelimiter("\\n")
  87 +// .setType(IotTcpCodecTypeEnum.LENGTH_FIELD.getType())
  88 +// .setLengthFieldOffset(0)
  89 +// .setLengthFieldLength(4)
  90 +// .setLengthAdjustment(0)
  91 +// .setInitialBytesToStrip(4)
  92 +// .setType(IotTcpCodecTypeEnum.LENGTH_FIELD.getType())
  93 +// .setFixedLength(256)
86 ); 94 );
87 95
88 // ===================== 网关设备信息(根据实际情况修改,从 iot_device 表查询网关设备) ===================== 96 // ===================== 网关设备信息(根据实际情况修改,从 iot_device 表查询网关设备) =====================
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotGatewaySubDeviceTcpProtocolIntegrationTest.java
@@ -7,7 +7,9 @@ import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; @@ -7,7 +7,9 @@ import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
7 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO; 7 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO;
8 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO; 8 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO;
9 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; 9 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  10 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpCodecTypeEnum;
10 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec; 11 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodec;
  12 +import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.codec.IotTcpFrameCodecFactory;
11 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer; 13 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
12 import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer; 14 import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer;
13 import io.vertx.core.Vertx; 15 import io.vertx.core.Vertx;
@@ -70,11 +72,17 @@ public class IotGatewaySubDeviceTcpProtocolIntegrationTest { @@ -70,11 +72,17 @@ public class IotGatewaySubDeviceTcpProtocolIntegrationTest {
70 /** 72 /**
71 * TCP 帧编解码器 73 * TCP 帧编解码器
72 */ 74 */
73 - private static final IotTcpFrameCodec FRAME_CODEC = IotTcpFrameCodec.create(  
74 - new IotTcpConfig.CodecConfig() {{  
75 - setType("delimiter");  
76 - setDelimiter("\\n");  
77 - }} 75 + private static final IotTcpFrameCodec FRAME_CODEC = IotTcpFrameCodecFactory.create(
  76 + new IotTcpConfig.CodecConfig()
  77 + .setType(IotTcpCodecTypeEnum.DELIMITER.getType())
  78 + .setDelimiter("\\n")
  79 +// .setType(IotTcpCodecTypeEnum.LENGTH_FIELD.getType())
  80 +// .setLengthFieldOffset(0)
  81 +// .setLengthFieldLength(4)
  82 +// .setLengthAdjustment(0)
  83 +// .setInitialBytesToStrip(4)
  84 +// .setType(IotTcpCodecTypeEnum.LENGTH_FIELD.getType())
  85 +// .setFixedLength(256)
78 ); 86 );
79 87
80 // ===================== 网关子设备信息(根据实际情况修改,从 iot_device 表查询子设备) ===================== 88 // ===================== 网关子设备信息(根据实际情况修改,从 iot_device 表查询子设备) =====================