+ * 执行当前编辑内容：不保存、不发消息、不改属性
+ *
+ * @author 芋道源码
+ */
+@Schema(description = "管理后台 - IoT 产品数据解析脚本调试 Request VO")
+@Data
+public class IotProductScriptDebugReqVO {
+
+ @Schema(description = "产品编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1024")
+ @NotNull(message = "产品编号不能为空")
+ private Long id;
+
+ @Schema(description = "数据解析脚本（JS ES5）", requiredMode = Schema.RequiredMode.REQUIRED)
+ @NotEmpty(message = "脚本内容不能为空")
+ private String scriptContent;
+
+ @Schema(description = "方向：decode 上行解码，encode 下行编码", requiredMode = Schema.RequiredMode.REQUIRED, example = "decode")
+ @NotEmpty(message = "调试方向不能为空")
+ @Pattern(regexp = "decode|encode", message = "调试方向只能是 decode 或 encode")
+ private String direction;
+
+ @Schema(description = "上行原始字节的 HEX（decode 时必填，允许空格）", example = "01 03 00 00")
+ private String inputHex;
+
+ @Schema(description = "下行指令（encode 时必填）：{method, requestId?, params?}")
+ private Map
+ * 部分更新：scriptContent、scriptEnabled 为 null 时不修改
+ *
+ * @author 芋道源码
+ */
+@Schema(description = "管理后台 - IoT 产品数据解析脚本更新 Request VO")
+@Data
+public class IotProductScriptUpdateReqVO {
+
+ @Schema(description = "产品编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1024")
+ @NotNull(message = "产品编号不能为空")
+ private Long id;
+
+ @Schema(description = "数据解析脚本（JS ES5）；空字符串表示清空，仅在停用时允许")
+ private String scriptContent;
+
+ @Schema(description = "是否启用数据解析脚本", example = "true")
+ private Boolean scriptEnabled;
+
+ @AssertTrue(message = "脚本内容和启用状态至少提交一个")
+ @JsonIgnore
+ public boolean isAnyFieldPresent() {
+ return scriptContent != null || scriptEnabled != null;
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceMessageDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceMessageDO.java
index 3b24aa9..5f81ffb 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceMessageDO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceMessageDO.java
@@ -106,4 +106,13 @@ public class IotDeviceMessageDO {
*/
private String msg;
+ // ========== 透传（raw）字段 ==========
+
+ /**
+ * 与设备之间实际收发的原始字节（Base64）
+ *
+ * 只由服务端可信赋值，见 {@link IotDeviceMessage#getRaw()}；为空表示没有原始帧
+ */
+ private String raw;
+
}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java
index a1a77fc..6536c63 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/product/IotProductDO.java
@@ -90,4 +90,13 @@ public class IotProductDO extends TenantBaseDO {
*/
private String serializeType;
+ /**
+ * 数据解析脚本（JS ES5），UTF-8 不超过 32KB
+ */
+ private String scriptContent;
+ /**
+ * 是否启用数据解析脚本
+ */
+ private Boolean scriptEnabled;
+
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBCodec.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBCodec.java
index 95afc90..eb13802 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBCodec.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBCodec.java
@@ -41,7 +41,8 @@ public class IotDeviceMessageIoTDBCodec {
.setParams(encodeString(toJsonString(message.getParams())))
.setData(encodeString(toJsonString(message.getData())))
.setMsg(encodeString(message.getMsg())).setReportTime(encodeString(message.getReportTime()))
- .setTenantId(encodeString(message.getTenantId())).setCode(encodeString(message.getCode()));
+ .setTenantId(encodeString(message.getTenantId())).setCode(encodeString(message.getCode()))
+ .setRaw(encodeString(message.getRaw()));
}
public static IotDeviceMessageDO decode(IotDeviceMessageIoTDBRow row) {
@@ -57,7 +58,8 @@ public class IotDeviceMessageIoTDBCodec {
.setMsg(decodeString(row.getMsg()))
.setReportTime(reportTime != null ? Long.valueOf(reportTime) : null)
.setTenantId(tenantId != null ? Long.valueOf(tenantId) : null)
- .setCode(code != null ? Integer.valueOf(code) : null);
+ .setCode(code != null ? Integer.valueOf(code) : null)
+ .setRaw(decodeString(row.getRaw()));
}
/**
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBDAO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBDAO.java
index 1af22d9..4ab796b 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBDAO.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/iotdb/IotDeviceMessageIoTDBDAO.java
@@ -50,6 +50,7 @@ public class IotDeviceMessageIoTDBDAO implements IotDeviceMessageDAO {
.put("method", "STRING").put("identifier", "STRING").put("request_id", "STRING").put("server_id", "STRING")
.put("params", "STRING").put("data", "STRING").put("msg", "STRING")
.put("report_time", "STRING").put("tenant_id", "STRING").put("code", "STRING")
+ .put("raw", "STRING")
.build();
@Resource
@@ -152,15 +153,20 @@ public class IotDeviceMessageIoTDBDAO implements IotDeviceMessageDAO {
Integer upstream = reqVO.getUpstream() != null ? encodeBoolean(reqVO.getUpstream()) : null;
Integer reply = reqVO.getReply() != null ? encodeBoolean(reqVO.getReply()) : null;
String identifier = StrUtil.isNotEmpty(reqVO.getIdentifier()) ? encodeString(reqVO.getIdentifier()) : null;
+ // 是否有原始帧：true 为 v: 开头；false 为 n:
+ Boolean raw = reqVO.getRaw();
+ String rawValuePattern = PREFIX_VALUE + "%";
try {
// 2.1 查询总数
- Long total = deviceMessageMapper.selectCount(deviceId, method, upstream, reply, identifier);
+ Long total = deviceMessageMapper.selectCount(deviceId, method, upstream, reply, identifier,
+ raw, PREFIX_NULL, rawValuePattern);
if (total == null || total == 0) {
return PageResult.empty();
}
// 2.2 查询列表：pageSize 为 -1 时，不分页
boolean pageable = ObjUtil.notEqual(reqVO.getPageSize(), PageParam.PAGE_SIZE_NONE);
List
+ * Rhino 下死循环会被预算中断、线程可复用；不经过检查点的原生长操作仍可能卡住线程 → 熔断调试功能，需要重启 biz
+ */
+ private IotScriptExecutor debugScriptExecutor;
+
+ @PostConstruct
+ public void initDebugScriptExecutor() {
+ debugScriptExecutor = new IotRhinoScriptExecutor(new IotRhinoScriptExecutor.Options().setName("iot-script-debug")
+ .setPoolSize(2).setQueueCapacity(4).setMaxStuckThreads(2));
+ }
+
+ @PreDestroy
+ public void destroyDebugScriptExecutor() {
+ debugScriptExecutor.shutdown();
+ }
+
@Override
public Long createProduct(IotProductSaveReqVO createReqVO) {
// 1. 校验 ProductKey
@@ -204,4 +262,157 @@ public class IotProductServiceImpl implements IotProductService {
}
}
+ // ========== 数据解析脚本 ==========
+
+ @Override
+ @CacheEvict(value = RedisKeyConstants.PRODUCT, key = "#updateReqVO.id")
+ public void updateProductScript(IotProductScriptUpdateReqVO updateReqVO) {
+ // 1.1 校验存在
+ IotProductDO product = validateProductExists(updateReqVO.getId());
+ // 1.2 校验脚本：UTF-8 不超过 32KB；语法只编译、不执行
+ if (updateReqVO.getScriptContent() != null) {
+ validateProductScript(updateReqVO.getScriptContent());
+ }
+ // 1.3 启用时内容不能为空：没提交的字段取库里的值
+ String content = ObjUtil.defaultIfNull(updateReqVO.getScriptContent(), product.getScriptContent());
+ Boolean enabled = ObjUtil.defaultIfNull(updateReqVO.getScriptEnabled(), product.getScriptEnabled());
+ if (Boolean.TRUE.equals(enabled) && StrUtil.isBlank(content)) {
+ throw exception(PRODUCT_SCRIPT_EMPTY);
+ }
+
+ // 2. 部分更新：只写本次提交的字段（null 跳过），保存内容不会顺带改启用状态
+ IotProductDO updateObj = BeanUtils.toBean(updateReqVO, IotProductDO.class);
+ productMapper.updateById(updateObj);
+ }
+
+ private void validateProductScriptLength(String content) {
+ if (StrUtil.utf8Bytes(content).length > SCRIPT_MAX_BYTES) {
+ throw exception(PRODUCT_SCRIPT_TOO_LONG, SCRIPT_MAX_BYTES);
+ }
+ }
+
+ private void validateProductScript(String content) {
+ validateProductScriptLength(content);
+ // 空字符串表示清空
+ if (StrUtil.isEmpty(content)) {
+ return;
+ }
+ // 在调试池线程上编译：只做语法检查，不执行任何用户代码
+ IotScriptResult result = debugScriptExecutor.compileOnly(content, COMPILE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ if (result.getStatus() == IotScriptResult.Status.REJECTED || result.isTimeout()) {
+ throw exception(PRODUCT_SCRIPT_DEBUG_BUSY);
+ }
+ if (!result.isSuccess()) {
+ throw exception(PRODUCT_SCRIPT_COMPILE_FAILED, result.getLineNumber() > 0
+ ? StrUtil.format("第 {} 行：{}", result.getLineNumber(), result.getMessage()) : result.getMessage());
+ }
+ }
+
+ @Override
+ public IotProductScriptDebugRespVO debugProductScript(IotProductScriptDebugReqVO debugReqVO) {
+ // 1.1 校验当前租户内的产品
+ IotProductDO product = validateProductExists(debugReqVO.getId());
+ // 1.2 校验脚本长度
+ validateProductScriptLength(debugReqVO.getScriptContent());
+ // 1.3 校验调试输入，构建 ctx
+ String ctx = buildDebugContext(debugReqVO, product);
+
+ // 2.1 执行：与网关相同的运行时和预算；请求线程阻塞等待，最长到截止时间
+ IotScriptResult result = debugScriptExecutor.execute(IotScriptSource.ofDebug(debugReqVO.getScriptContent()),
+ debugReqVO.getDirection(), ctx, System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DEBUG_DEADLINE_MILLIS));
+ // 2.2 构建结果：与网关相同的输出校验
+ return buildDebugResult(debugReqVO, product, result);
+ }
+
+ /**
+ * 构建调试的 ctx：与网关的结构一致，但设备名固定为 {@link #DEBUG_DEVICE_NAME}，不带任何真实设备的信息或秘密
+ *
+ * @param reqVO 调试请求
+ * @param product 产品
+ * @return ctx（JSON）
+ */
+ private static String buildDebugContext(IotProductScriptDebugReqVO reqVO, IotProductDO product) {
+ // 1. decode：上行原始字节
+ if (IotScriptUtils.FUNCTION_DECODE.equals(reqVO.getDirection())) {
+ // 1.1 校验 HEX（允许空格）
+ String hex = StrUtil.cleanBlank(reqVO.getInputHex());
+ if (StrUtil.isEmpty(hex) || hex.length() % 2 != 0 || !ReUtil.isMatch("[0-9a-fA-F]+", hex)) {
+ throw invalidParamException("上行数据必须是偶数位的 HEX");
+ }
+ // 1.2 校验长度：与透传上报的上限一致
+ byte[] bytes = HexUtil.decodeHex(hex);
+ if (bytes.length > IotDeviceMessageUtils.RAW_PAYLOAD_MAX_BYTES) {
+ throw invalidParamException("上行数据不能超过 {} 字节", IotDeviceMessageUtils.RAW_PAYLOAD_MAX_BYTES);
+ }
+ // 1.3 构建 decode 的 ctx
+ return IotScriptUtils.buildDecodeContext(product.getProtocolType(), product.getProductKey(),
+ DEBUG_DEVICE_NAME, System.currentTimeMillis(), bytes, null);
+ }
+
+ // 2.1 encode：与网关一致，只编码 property.set / service.invoke 请求
+ Map
+ * 目前只有 Rhino 实现 {@link cn.iocoder.yudao.module.iot.core.script.rhino.IotRhinoScriptExecutor}，由使用方直接创建 + * + * @author 芋道源码 + */ +public interface IotScriptExecutor { + + /** + * 执行一次脚本调用：调用方线程阻塞等待，最长到截止时间 + * + * @param source 脚本 + * @param function "decode" 或 "encode" + * @param ctxJson 平台构造的入参 JSON + * @param deadlineNanos 截止时间（System.nanoTime）：覆盖排队与执行 + * @return 结果，状态见 {@link IotScriptResult.Status} + */ + IotScriptResult execute(IotScriptSource source, String function, String ctxJson, long deadlineNanos); + + /** + * 只做语法检查（不执行任何用户代码）：调用方线程有界等待 + * + * @param content 脚本内容 + * @param timeout 等待时间 + * @param unit 时间单位 + * @return 结果：SUCCESS / ERROR（带行号）/ REJECTED / TIMEOUT + */ + IotScriptResult compileOnly(String content, long timeout, TimeUnit unit); + + /** + * 预热：启动时、接收流量之前调用一次 + */ + void warmUp(); + + /** + * 停止：不再接收新任务 + */ + void shutdown(); + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/IotScriptUtils.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/IotScriptUtils.java new file mode 100644 index 0000000..2952825 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/IotScriptUtils.java @@ -0,0 +1,344 @@ +package cn.iocoder.yudao.module.iot.core.script; + +import cn.hutool.core.util.HexUtil; +import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptResult; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import com.fasterxml.jackson.databind.JsonNode; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.IntStream; + +/** + * IoT 数据解析脚本的契约：入参构造与输出校验 + *
+ * 网关所有协议和 biz 调试共用同一份规则，结果一致。任何一项不合法就整体拒绝（{@link IotScriptResult.Status#OUTPUT_INVALID}），不截断、不部分写出
+ *
+ * @author 芋道源码
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class IotScriptUtils {
+
+ public static final String FUNCTION_DECODE = "decode";
+ public static final String FUNCTION_ENCODE = "encode";
+
+ /**
+ * 消息 params / data 序列化后的最大 UTF-8 字节数
+ */
+ public static final int MAX_FIELD_BYTES = 8192;
+ public static final int MAX_REQUEST_ID_LENGTH = 50;
+ public static final int MAX_IDENTIFIER_LENGTH = 100;
+ public static final int MAX_MSG_LENGTH = 256;
+
+ private static final Set
+ * 网关的 encode、biz 的脚本调试共用；raw.push 不经过 encode，由调用方自己校验
+ *
+ * @param json 脚本输出；null 表示没有定义 encode，或返回了 null / undefined
+ * @param protocol 协议类型：透传字节按它的规则校验
+ * @return null 表示没有生成下行数据（调用方只记日志，不写设备）
+ * @throws IotScriptException 输出不合法
+ */
+ public static EncodeOutput validateEncode(String json, IotProtocolTypeEnum protocol) throws IotScriptException {
+ // 1. 没有生成下行数据
+ if (json == null) {
+ return null;
+ }
+ // 2. 解析对象：只允许 bytes、standard，且只能返回一个
+ JsonNode root = parseObject(json, TOP_ENCODE_KEYS);
+ JsonNode standard = root.get("standard");
+ JsonNode bytesNode = root.get("bytes");
+ if (standard != null && bytesNode != null) {
+ throw IotScriptException.outputInvalid("bytes 和 standard 只能返回一个");
+ }
+ // 3.1 走现有的标准路径
+ if (standard != null) {
+ if (!standard.isBoolean() || !standard.booleanValue()) {
+ throw IotScriptException.outputInvalid("standard 只能是 true");
+ }
+ return new EncodeOutput(true, null);
+ }
+ // 3.2 透传字节：按协议的规则校验
+ byte[] bytes = parseBytes(bytesNode, "bytes");
+ try {
+ IotDeviceMessageUtils.validateRawPayload(protocol, bytes);
+ } catch (IllegalArgumentException e) {
+ throw IotScriptException.outputInvalid(e.getMessage());
+ }
+ return new EncodeOutput(false, bytes);
+ }
+
+ private static JsonNode parseObject(String json, Set
+ * core 不依赖 biz / gateway 的错误码：由调用方按 {@link Status} 翻译为各自的 ErrorCode
+ *
+ * @author 芋道源码
+ */
+@Getter
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public final class IotScriptResult {
+
+ /**
+ * 调用结果的状态
+ */
+ public enum Status {
+
+ /**
+ * 执行成功（json 可能为 null：没有定义该函数，或返回了 null / undefined）
+ */
+ SUCCESS,
+ /**
+ * 脚本异常：语法错误、运行错误、throw、超出栈深等
+ */
+ ERROR,
+ /**
+ * 超时：排队超过截止时间、超过执行预算（时间或指令）
+ */
+ TIMEOUT,
+ /**
+ * 拒绝：JS 线程池已满、已熔断
+ */
+ REJECTED,
+ /**
+ * 输出不合法
+ */
+ OUTPUT_INVALID
+
+ }
+
+ // DONE @AI：字段注释，还是要写下
+ /**
+ * 状态
+ */
+ private final Status status;
+ /**
+ * 结果 JSON；只有 SUCCESS 时可能非空
+ */
+ private final String json;
+ /**
+ * 错误信息
+ */
+ private final String message;
+ /**
+ * 出错行号；未知时为 -1
+ */
+ private final int lineNumber;
+ /**
+ * 执行耗时（毫秒）
+ */
+ private final long durationMillis;
+
+ public static IotScriptResult success(String json, long durationMillis) {
+ return new IotScriptResult(Status.SUCCESS, json, null, -1, durationMillis);
+ }
+
+ public static IotScriptResult error(String message, long durationMillis) {
+ return new IotScriptResult(Status.ERROR, null, message, -1, durationMillis);
+ }
+
+ public static IotScriptResult timeout(String message) {
+ return new IotScriptResult(Status.TIMEOUT, null, message, -1, 0);
+ }
+
+ public static IotScriptResult rejected(String message) {
+ return new IotScriptResult(Status.REJECTED, null, message, -1, 0);
+ }
+
+ public static IotScriptResult of(Status status, String message, int lineNumber, long durationMillis) {
+ return new IotScriptResult(status, null, message, lineNumber, durationMillis);
+ }
+
+ public boolean isSuccess() {
+ return status == Status.SUCCESS;
+ }
+
+ public boolean isTimeout() {
+ return status == Status.TIMEOUT;
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/dto/IotScriptSource.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/dto/IotScriptSource.java
new file mode 100644
index 0000000..8d96879
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/dto/IotScriptSource.java
@@ -0,0 +1,56 @@
+package cn.iocoder.yudao.module.iot.core.script.dto;
+
+import cn.hutool.crypto.digest.DigestUtil;
+import lombok.Getter;
+
+/**
+ * IoT 数据解析脚本：不可变对象 {productId, hash, content}
+ *
+ * hash = 源码 UTF-8 的完整 SHA-256（64 位 hex），作为缓存和编译身份；{@link #getShortHash()} 只用于日志和界面。
+ * 源码字符串可以被多个线程只读共享，编译结果不共享
+ *
+ * @author 芋道源码
+ */
+@Getter
+public final class IotScriptSource {
+
+ /**
+ * 产品编号；调试时为 null
+ */
+ private final Long productId;
+ /**
+ * 源码 UTF-8 的完整 SHA-256
+ */
+ private final String hash;
+ /**
+ * 源码
+ */
+ private final String content;
+ /**
+ * 编译缓存的 key：productId:hash；调试时为 debug:hash
+ */
+ private final String key;
+
+ private IotScriptSource(Long productId, String content) {
+ this.productId = productId;
+ this.content = content;
+ this.hash = DigestUtil.sha256Hex(content);
+ this.key = (productId != null ? productId.toString() : "debug") + ":" + hash;
+ }
+
+ public static IotScriptSource of(Long productId, String content) {
+ return new IotScriptSource(productId, content);
+ }
+
+ public static IotScriptSource ofDebug(String content) {
+ return new IotScriptSource(null, content);
+ }
+
+ /**
+ * @return hash 前 12 位：只用于日志和界面，不作为唯一标识
+ */
+ public String getShortHash() {
+ return hash.substring(0, 12);
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/package-info.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/package-info.java
new file mode 100644
index 0000000..25b94c8
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/package-info.java
@@ -0,0 +1,13 @@
+/**
+ * IoT 数据解析脚本：网关所有协议与 biz 调试共用
+ *
+ * Rhino 下，经过指令检查点的超限会真正中断执行、释放线程；不经过检查点的原生长操作只能被标记为超时，线程计入卡住检测。
+ * 本节点的熔断（按 productId:hash，或卡住线程过多时全局）只在内存中生效，不写库，重启后复位
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotRhinoScriptExecutor implements IotScriptExecutor {
+
+ /**
+ * 熔断：1 分钟内超时达到该次数
+ */
+ private static final int BREAKER_TIMEOUT_COUNT = 3;
+ // DONE @AI：字段注释；
+ /**
+ * 熔断：统计超时次数的时间窗口（1 分钟）
+ */
+ private static final long BREAKER_TIMEOUT_WINDOW_NANOS = TimeUnit.MINUTES.toNanos(1);
+
+ // DONE @AI：字段注释；
+ /**
+ * IoT 专用的 Rhino ContextFactory：所有 JS 线程共用，不使用全局 ContextFactory
+ */
+ private final IotScriptContextFactory contextFactory = new IotScriptContextFactory();
+ // DONE @AI：字段注释；
+ /**
+ * 每个 JS 线程独占的 Rhino 运行时：预热或首次调用时创建，不跨线程共享
+ */
+ private final ThreadLocal
+ * 脚本在 JS 线程池执行，受执行预算（时间、指令）约束；等待到期时，还在排队的任务被终结，之后不会再启动
+ *
+ * @param source 脚本
+ * @param function "decode" 或 "encode"
+ * @param ctxJson 平台构造的入参 JSON
+ * @param deadlineNanos 截止时间（System.nanoTime）：覆盖排队与执行
+ * @return 结果：SUCCESS / ERROR / TIMEOUT / REJECTED
+ */
+ @Override
+ public IotScriptResult execute(IotScriptSource source, String function, String ctxJson, long deadlineNanos) {
+ if (isOpen(source)) {
+ return IotScriptResult.rejected("脚本已熔断");
+ }
+ long remain = deadlineNanos - System.nanoTime();
+ if (remain <= 0) {
+ return IotScriptResult.timeout("进入 JS 池前已超过截止时间");
+ }
+ IotScriptTask task = new IotScriptTask(source);
+ try {
+ jsPool.execute(() -> run(task, function, ctxJson, deadlineNanos));
+ } catch (RejectedExecutionException e) {
+ return IotScriptResult.rejected("脚本线程池已满");
+ }
+ try {
+ return task.getResult().get(remain, TimeUnit.NANOSECONDS);
+ } catch (TimeoutException e) {
+ // 到期：排队中 PENDING → DONE（结果为超时），之后 markRunning 失败，不会启动；执行中的由预算中断或计入卡住检测
+ finish(task, IotScriptResult.timeout("超过截止时间"));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ finish(task, IotScriptResult.error("等待脚本结果被中断", 0));
+ } catch (ExecutionException e) {
+ finish(task, IotScriptResult.error(StrUtil.maxLength(String.valueOf(e.getCause()), 200), 0));
+ }
+ return task.getResult().join(); // finish 只会成功一次：这里拿到的是最终终结的那个结果
+ }
+
+ /**
+ * 只做语法检查（不执行任何用户代码）：在 JS 线程上编译，调用方线程有界等待
+ *
+ * @param content 脚本内容
+ * @param timeout 等待时间
+ * @param unit 时间单位
+ * @return 结果：SUCCESS / ERROR（带行号）/ REJECTED / TIMEOUT
+ */
+ @Override
+ public IotScriptResult compileOnly(String content, long timeout, TimeUnit unit) {
+ if (isGlobalOpen()) {
+ return IotScriptResult.rejected("脚本已熔断");
+ }
+ Future
+ * 失败 → 全局熔断（记 ERROR），之后的调用都按「已熔断」拒绝
+ */
+ @Override
+ public void warmUp() {
+ int n = jsPool.getCorePoolSize();
+ try {
+ jsPool.prestartAllCoreThreads();
+ CyclicBarrier barrier = new CyclicBarrier(n);
+ List
+ * 只代表「这一次脚本调用」的终态。终结时取消的是本任务自己的预算计时器，不涉及调用方（协议）的响应计时器
+ */
+ static final class IotScriptTask {
+
+ /**
+ * 状态：排队中，还没开始
+ */
+ static final int PENDING = 0;
+ /**
+ * 状态：JS 线程执行中
+ */
+ static final int RUNNING = 1;
+ /**
+ * 状态：已终结（成功、错误、拒绝或超时；具体原因见终结的结果）
+ */
+ static final int DONE = 2;
+
+ /**
+ * 当前状态：PENDING → RUNNING（markRunning）；PENDING 或 RUNNING → DONE（finish），只能终结一次
+ */
+ private final AtomicInteger state = new AtomicInteger(PENDING);
+ /**
+ * 本次调用的脚本
+ */
+ @Getter
+ private final IotScriptSource source;
+ /**
+ * 终结的结果：调用方在 {@link #execute} 中等待
+ */
+ @Getter
+ private final CompletableFuture
+ * 只在本次调用的 JS 线程上使用，不需要同步
+ *
+ * @author 芋道源码
+ */
+final class IotScriptBudget {
+
+ /**
+ * 截止时间（System.nanoTime）= min(开始执行 + execBudget, 调用方传入的截止时间)
+ */
+ private final long deadlineNanos;
+ /**
+ * 指令上限
+ */
+ private final long maxInstructions;
+ /**
+ * 已执行的指令数
+ */
+ private long usedInstructions;
+
+ IotScriptBudget(long deadlineNanos, long maxInstructions) {
+ this.deadlineNanos = deadlineNanos;
+ this.maxInstructions = maxInstructions;
+ }
+
+ void consume(int count) {
+ usedInstructions += count;
+ if (usedInstructions >= maxInstructions) {
+ throw new BudgetExceededError("超过指令预算");
+ }
+ if (System.nanoTime() >= deadlineNanos) {
+ throw new BudgetExceededError("超过执行时间预算");
+ }
+ }
+
+ /**
+ * 观察器抛出的宿主 Error：Rhino 原样向上抛，脚本的 catch/finally 拦不住
+ *
+ * 一旦超限，之后每个检查点都会再抛，所以 finally 中的死循环也会被中断
+ */
+ static final class BudgetExceededError extends Error {
+
+ BudgetExceededError(String message) {
+ super(message, null, false, false); // 不收集栈：开销小
+ }
+
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotScriptContextFactory.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotScriptContextFactory.java
new file mode 100644
index 0000000..fff0a26
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotScriptContextFactory.java
@@ -0,0 +1,43 @@
+package cn.iocoder.yudao.module.iot.core.script.rhino;
+
+import org.mozilla.javascript.Context;
+import org.mozilla.javascript.ContextFactory;
+
+/**
+ * IoT 脚本专用的 Rhino ContextFactory：不使用 ContextFactory.getGlobal()，不影响进程内其它 Rhino 使用方
+ *
+ * @author 芋道源码
+ */
+final class IotScriptContextFactory extends ContextFactory {
+
+ /**
+ * 每执行这么多条解释器指令，回调一次 observeInstructionCount
+ */
+ static final int INSTRUCTION_THRESHOLD = 1000;
+ /**
+ * 解释器最大栈深（递归深度）
+ */
+ static final int MAX_STACK_DEPTH = 64;
+
+ @Override
+ protected Context makeContext() {
+ Context cx = super.makeContext();
+ cx.setOptimizationLevel(-1); // 解释模式：指令观察器只在解释模式下生效
+ cx.setLanguageVersion(Context.VERSION_1_8);
+ cx.setInstructionObserverThreshold(INSTRUCTION_THRESHOLD);
+ cx.setMaximumInterpreterStackDepth(MAX_STACK_DEPTH); // 超出 → EvaluatorException「Exceeded maximum stack depth」
+ cx.setClassShutter(className -> false); // 拒绝任何 Java 类
+ cx.setGeneratingDebug(false);
+ return cx;
+ }
+
+ @Override
+ protected void observeInstructionCount(Context cx, int instructionCount) {
+ IotScriptBudget budget = (IotScriptBudget) cx.getThreadLocal(IotScriptBudget.class);
+ if (budget == null) {
+ throw new IotScriptBudget.BudgetExceededError("缺少执行预算"); // fail-closed：没有预算的 Context 不允许继续执行
+ }
+ budget.consume(instructionCount); // 超限时抛 BudgetExceededError
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotScriptRuntime.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotScriptRuntime.java
new file mode 100644
index 0000000..78c5a8a
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotScriptRuntime.java
@@ -0,0 +1,179 @@
+package cn.iocoder.yudao.module.iot.core.script.rhino;
+
+import cn.hutool.core.util.StrUtil;
+import cn.iocoder.yudao.module.iot.core.script.IotScriptException;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptResult;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptSource;
+import org.mozilla.javascript.Context;
+import org.mozilla.javascript.Function;
+import org.mozilla.javascript.RhinoException;
+import org.mozilla.javascript.Script;
+import org.mozilla.javascript.Scriptable;
+import org.mozilla.javascript.ScriptableObject;
+import org.mozilla.javascript.Undefined;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+// DONE @AI：字段注释；
+/**
+ * 单个 JS worker 线程独占的 Rhino 运行时
+ *
+ * biz 的 /send、脚本调试的 encode 输出、gateway 的 encode 输出调用同一份规则；不合法抛 {@link IllegalArgumentException}
+ *
+ * @param protocol 协议类型
+ * @param bytes 下行字节
+ */
+ public static void validateRawPayload(IotProtocolTypeEnum protocol, byte[] bytes) {
+ if (protocol != null && protocol.isModbus()) {
+ validateModbusRequestPdu(bytes);
+ return;
+ }
+ Assert.isTrue(bytes != null && bytes.length >= 1 && bytes.length <= RAW_PAYLOAD_MAX_BYTES,
+ "raw 长度必须在 1~{} 字节之间", RAW_PAYLOAD_MAX_BYTES);
+ }
+
+ /**
+ * 校验 Modbus 原始请求 PDU 的格式：功能码白名单、长度、数量范围、05 的取值、0F/10 的 byteCount
+ *
+ * 透传下行的字节规则（{@link #validateRawPayload}）与 Modbus 协议写出前调用同一份规则；不合法抛 {@link IllegalArgumentException}
+ *
+ * @param pdu PDU（功能码 + 数据，不带 slaveId、CRC、MBAP）
+ */
+ public static void validateModbusRequestPdu(byte[] pdu) {
+ Assert.isTrue(pdu != null && pdu.length >= 5 && pdu.length <= MODBUS_PDU_MAX_BYTES,
+ "PDU 长度必须在 5~{} 字节之间", MODBUS_PDU_MAX_BYTES);
+ int functionCode = pdu[0] & 0xFF;
+ int quantity = ((pdu[3] & 0xFF) << 8) | (pdu[4] & 0xFF);
+ switch (functionCode) {
+ case 0x01:
+ case 0x02:
+ Assert.isTrue(pdu.length == 5 && quantity >= 1 && quantity <= 2000, "PDU 不合法：读线圈数量必须在 1~2000 之间");
+ break;
+ case 0x03:
+ case 0x04:
+ Assert.isTrue(pdu.length == 5 && quantity >= 1 && quantity <= 125, "PDU 不合法：读寄存器数量必须在 1~125 之间");
+ break;
+ case 0x05:
+ Assert.isTrue(pdu.length == 5 && (quantity == 0xFF00 || quantity == 0x0000), "PDU 不合法：写单个线圈的值只能是 FF00 或 0000");
+ break;
+ case 0x06:
+ Assert.isTrue(pdu.length == 5, "PDU 不合法：写单个寄存器必须是 5 字节");
+ break;
+ case 0x0F:
+ Assert.isTrue(quantity >= 1 && quantity <= 1968 && pdu.length >= 6 && pdu.length == 6 + (pdu[5] & 0xFF)
+ && (pdu[5] & 0xFF) == (quantity + 7) / 8, "PDU 不合法：写多个线圈的数量或字节数不一致");
+ break;
+ case 0x10:
+ Assert.isTrue(quantity >= 1 && quantity <= 123 && pdu.length >= 6 && pdu.length == 6 + (pdu[5] & 0xFF)
+ && (pdu[5] & 0xFF) == quantity * 2, "PDU 不合法：写多个寄存器的数量或字节数不一致");
+ break;
+ default:
+ throw new IllegalArgumentException("PDU 不合法：不支持的功能码 " + functionCode);
+ }
+ }
+
+ /**
* 提取消息中的标识符
*
* @param message 消息
diff --git a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/script/IotScriptUtilsTest.java b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/script/IotScriptUtilsTest.java
new file mode 100644
index 0000000..6230d26
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/script/IotScriptUtilsTest.java
@@ -0,0 +1,133 @@
+package cn.iocoder.yudao.module.iot.core.script;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptResult;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@link IotScriptUtils} 的单元测试
+ *
+ * @author 芋道源码
+ */
+public class IotScriptUtilsTest {
+
+ @Test
+ public void testValidateDecode_success() throws IotScriptException {
+ // 调用：一个标准消息对象
+ IotDeviceMessage message = IotScriptUtils.validateDecode(
+ "{\"method\":\"thing.property.post\",\"params\":{\"t\":25}}", null);
+ // 断言
+ assertEquals("thing.property.post", message.getMethod());
+ assertEquals(25, ((Map) message.getParams()).get("t"));
+ assertNull(message.getCode());
+ assertNull(message.getRaw());
+ // 调用 + 断言：事件
+ IotDeviceMessage event = IotScriptUtils.validateDecode(
+ "{\"method\":\"thing.event.post\",\"params\":{\"identifier\":\"alarm\",\"value\":{}}}", null);
+ assertEquals("thing.event.post", event.getMethod());
+ }
+
+ @Test
+ public void testValidateDecode_null() throws IotScriptException {
+ // 调用 + 断言：decode 未定义或返回 null → 没有业务消息
+ assertNull(IotScriptUtils.validateDecode(null, null));
+ }
+
+ @Test
+ public void testValidateDecode_invalid() {
+ // 不是对象、旧的 messages / ack 包装
+ assertOutputInvalid("[]");
+ assertOutputInvalid("{\"messages\":[{\"method\":\"thing.property.post\",\"params\":{}}]}");
+ assertOutputInvalid("{\"method\":\"thing.property.post\",\"params\":{},\"ack\":[1]}");
+ // 身份、路由、可信字段和未知键不允许：脚本不能伪造 raw、设备、租户、编号
+ assertOutputInvalid("{\"method\":\"thing.property.post\",\"params\":{},\"deviceId\":1}");
+ assertOutputInvalid("{\"method\":\"thing.property.post\",\"params\":{},\"tenantId\":1}");
+ assertOutputInvalid("{\"method\":\"thing.property.post\",\"params\":{},\"id\":\"x\"}");
+ assertOutputInvalid("{\"method\":\"thing.property.post\",\"params\":{},\"raw\":\"AQI=\"}");
+ assertOutputInvalid("{\"method\":\"thing.property.post\",\"params\":{},\"unknown\":true}");
+ // 不支持的方法；事件缺 identifier；回执缺 requestId
+ assertOutputInvalid("{\"method\":\"thing.raw.post\",\"params\":{}}");
+ assertOutputInvalid("{\"method\":\"thing.event.post\",\"params\":{}}");
+ assertOutputInvalid("{\"method\":\"thing.property.set\",\"code\":0}");
+ }
+
+ @Test
+ public void testValidateDecode_transactionReply() throws IotScriptException {
+ // 准备参数
+ IotDeviceMessage request = IotDeviceMessage.requestOf("req-1", "thing.property.set", null);
+
+ // 调用 + 断言：方法、requestId 一致的回执
+ IotDeviceMessage reply = IotScriptUtils.validateDecode(
+ "{\"method\":\"thing.property.set\",\"requestId\":\"req-1\",\"code\":0}", request);
+ assertEquals(0, reply.getCode());
+ // 调用 + 断言：requestId 不一致、不是回执、没有输出
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateDecode(
+ "{\"method\":\"thing.property.set\",\"requestId\":\"req-2\",\"code\":0}", request));
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateDecode(
+ "{\"method\":\"thing.property.post\",\"params\":{}}", request));
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateDecode(null, request));
+ }
+
+ @Test
+ public void testValidateEncode() throws IotScriptException {
+ // 未定义或返回 null → 不写设备
+ assertNull(IotScriptUtils.validateEncode(null, IotProtocolTypeEnum.TCP));
+ // 走标准路径
+ assertTrue(IotScriptUtils.validateEncode("{\"standard\":true}", IotProtocolTypeEnum.TCP).isStandard());
+ // 字节
+ IotScriptUtils.EncodeOutput output = IotScriptUtils.validateEncode("{\"bytes\":[1,2,255]}", IotProtocolTypeEnum.TCP);
+ assertFalse(output.isStandard());
+ assertArrayEquals(new byte[]{1, 2, (byte) 0xFF}, output.getBytes());
+ assertEquals(4096, IotScriptUtils.validateEncode("{\"bytes\":" + buildArray(4096, 0) + "}",
+ IotProtocolTypeEnum.TCP).getBytes().length);
+ // 不合法：standard=false、两者都有、没有 bytes、不是数组、超过 255
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateEncode("{\"standard\":false}", IotProtocolTypeEnum.TCP));
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateEncode("{\"standard\":true,\"bytes\":[1]}", IotProtocolTypeEnum.TCP));
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateEncode("{}", IotProtocolTypeEnum.TCP));
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateEncode("{\"bytes\":\"AQI=\"}", IotProtocolTypeEnum.TCP));
+ assertThrows(IotScriptException.class, () -> IotScriptUtils.validateEncode("{\"bytes\":[256]}", IotProtocolTypeEnum.TCP));
+ }
+
+ @Test
+ public void testValidateEncode_rawPayload() throws IotScriptException {
+ // 调用 + 断言：其它协议 1~4096 字节，不合法为 OUTPUT_INVALID，带原规则的错误信息
+ assertEncodeOutputInvalid("{\"bytes\":[]}", IotProtocolTypeEnum.TCP, "raw 长度必须在 1~4096 字节之间");
+ assertEncodeOutputInvalid("{\"bytes\":" + buildArray(4097, 0) + "}", IotProtocolTypeEnum.TCP,
+ "raw 长度必须在 1~4096 字节之间");
+
+ // 调用 + 断言：Modbus 必须是合法的请求 PDU
+ assertArrayEquals(new byte[]{3, 0, 0, 0, 10}, IotScriptUtils.validateEncode("{\"bytes\":[3,0,0,0,10]}",
+ IotProtocolTypeEnum.MODBUS_TCP_CLIENT).getBytes());
+ assertEncodeOutputInvalid("{\"bytes\":[3,0,0,0,0]}", IotProtocolTypeEnum.MODBUS_TCP_SERVER,
+ "PDU 不合法：读寄存器数量必须在 1~125 之间");
+ // 调用 + 断言：走标准路径时没有字节，不按协议的规则校验
+ assertTrue(IotScriptUtils.validateEncode("{\"standard\":true}", IotProtocolTypeEnum.MODBUS_TCP_CLIENT).isStandard());
+ }
+
+ private static void assertOutputInvalid(String json) {
+ IotScriptException ex = assertThrows(IotScriptException.class,
+ () -> IotScriptUtils.validateDecode(json, null), json);
+ assertEquals(IotScriptResult.Status.OUTPUT_INVALID, ex.getStatus());
+ }
+
+ private static void assertEncodeOutputInvalid(String json, IotProtocolTypeEnum protocol, String message) {
+ IotScriptException ex = assertThrows(IotScriptException.class,
+ () -> IotScriptUtils.validateEncode(json, protocol));
+ assertEquals(IotScriptResult.Status.OUTPUT_INVALID, ex.getStatus());
+ assertEquals(message, ex.getMessage());
+ }
+
+ private static String buildArray(int size, int value) {
+ StringBuilder sb = new StringBuilder("[");
+ for (int i = 0; i < size; i++) {
+ sb.append(i > 0 ? "," : "").append(value);
+ }
+ return sb.append("]").toString();
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotRhinoScriptExecutorTest.java b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotRhinoScriptExecutorTest.java
new file mode 100644
index 0000000..0633578
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotRhinoScriptExecutorTest.java
@@ -0,0 +1,203 @@
+package cn.iocoder.yudao.module.iot.core.script.rhino;
+
+import cn.hutool.core.util.ReflectUtil;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.script.IotScriptUtils;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptResult;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptSource;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@link IotRhinoScriptExecutor} 的单元测试：真实的 Rhino 解释执行
+ *
+ * @author 芋道源码
+ */
+public class IotRhinoScriptExecutorTest {
+
+ private static final String CTX = "{\"bytes\":[1,2,3],\"hex\":\"010203\"}";
+
+ private IotRhinoScriptExecutor executor;
+
+ @BeforeEach
+ public void setUp() {
+ executor = new IotRhinoScriptExecutor(new IotRhinoScriptExecutor.Options().setName("test-script-js").setPoolSize(1)
+ .setQueueCapacity(4).setExecBudgetMillis(500).setMaxInstructions(1_000_000));
+ }
+
+ @AfterEach
+ public void tearDown() {
+ executor.shutdown();
+ }
+
+ private IotScriptResult invoke(String content, String function) {
+ return executor.execute(IotScriptSource.of(1L, content), function, CTX,
+ System.nanoTime() + TimeUnit.SECONDS.toNanos(3));
+ }
+
+ @Test
+ public void testExecute_decodeSuccess() {
+ // 准备参数：宿主侧 JSON.parse 入参，函数返回原生对象，宿主侧 JSON.stringify
+ String content = "function decode(ctx) { return {method: 'thing.property.post',"
+ + " params: {sum: ctx.bytes[0] + ctx.bytes[1] + ctx.bytes[2], hex: ctx.hex}}; }";
+
+ // 调用
+ IotScriptResult result = invoke(content, IotScriptUtils.FUNCTION_DECODE);
+ // 断言
+ assertEquals(IotScriptResult.Status.SUCCESS, result.getStatus());
+ JsonNode root = JsonUtils.parseTree(result.getJson());
+ assertEquals(6, root.get("params").get("sum").intValue());
+ assertEquals("010203", root.get("params").get("hex").textValue());
+ }
+
+ @Test
+ public void testExecute_functionMissing() {
+ // 调用：没有定义 encode
+ IotScriptResult result = invoke("function decode(ctx) { return null; }", IotScriptUtils.FUNCTION_ENCODE);
+ // 断言：成功但没有输出，由调用方处理
+ assertEquals(IotScriptResult.Status.SUCCESS, result.getStatus());
+ assertNull(result.getJson());
+ }
+
+ @Test
+ public void testExecute_javaAccessDenied() {
+ // 调用：java、Packages、getClass 都不可用
+ String content = "function decode(ctx) { return {method: 'thing.property.post', params: {"
+ + "java: typeof java === 'undefined' ? 1 : 0, packages: typeof Packages === 'undefined' ? 1 : 0}}; }";
+ IotScriptResult result = invoke(content, IotScriptUtils.FUNCTION_DECODE);
+ // 断言
+ assertEquals(IotScriptResult.Status.SUCCESS, result.getStatus());
+ assertEquals("{\"java\":1,\"packages\":1}", JsonUtils.parseTree(result.getJson()).get("params").toString());
+
+ // 调用：显式访问 Java 类
+ IotScriptResult result2 = invoke("function decode(ctx) { return {method: 'thing.property.post',"
+ + " params: {home: java.lang.System.getProperty('user.home').length}}; }", IotScriptUtils.FUNCTION_DECODE);
+ // 断言：脚本错误，不是成功
+ assertEquals(IotScriptResult.Status.ERROR, result2.getStatus());
+ }
+
+ @Test
+ public void testExecute_infiniteLoopTimeoutAndReuse() {
+ // 调用：死循环被预算中断
+ IotScriptResult result = invoke("function decode(ctx) { while (true) {} }", IotScriptUtils.FUNCTION_DECODE);
+ // 断言
+ assertEquals(IotScriptResult.Status.TIMEOUT, result.getStatus());
+
+ // 调用：同一个线程继续可用（线程池只有 1 个线程）
+ IotScriptResult result2 = invoke("function decode(ctx) { return null; }", IotScriptUtils.FUNCTION_DECODE);
+ // 断言
+ assertEquals(IotScriptResult.Status.SUCCESS, result2.getStatus());
+ }
+
+ @Test
+ public void testExecute_breakerOpenAfterTimeouts() {
+ // 准备参数：同一个脚本 1 分钟内超时 3 次
+ String content = "function decode(ctx) { for (;;) {} }";
+ for (int i = 0; i < 3; i++) {
+ assertEquals(IotScriptResult.Status.TIMEOUT, invoke(content, IotScriptUtils.FUNCTION_DECODE).getStatus());
+ }
+
+ // 调用 + 断言：之后直接拒绝；其它脚本不受影响
+ assertTrue(executor.isOpen(IotScriptSource.of(1L, content)));
+ assertEquals(IotScriptResult.Status.REJECTED, invoke(content, IotScriptUtils.FUNCTION_DECODE).getStatus());
+ assertEquals(IotScriptResult.Status.SUCCESS, invoke("function decode(ctx) { return null; }",
+ IotScriptUtils.FUNCTION_DECODE).getStatus());
+ }
+
+ @Test
+ public void testExecute_deadlinePassed() {
+ // 调用：截止时间已过
+ IotScriptResult result = executor.execute(IotScriptSource.of(1L, "function decode(ctx) { return null; }"),
+ IotScriptUtils.FUNCTION_DECODE, CTX, System.nanoTime() - 1);
+ // 断言：不进入 JS 池
+ assertEquals(IotScriptResult.Status.TIMEOUT, result.getStatus());
+ }
+
+ @Test
+ public void testExecute_pendingTimeoutNeverStarts() throws Exception {
+ // 准备参数：用一个 Java 任务确定地占住唯一的 JS 线程（不依赖脚本预算与睡眠）；
+ // 该线程还没有执行过脚本（没有预热），所以它的 runtime 为空：任何脚本一旦在它上面启动，都会创建 runtime
+ ThreadPoolExecutor jsPool = (ThreadPoolExecutor) ReflectUtil.getFieldValue(executor, "jsPool");
+ ThreadLocal runtimes = (ThreadLocal) ReflectUtil.getFieldValue(executor, "runtimes");
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ jsPool.execute(() -> {
+ entered.countDown();
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+
+ // 调用：排队中的任务 100ms 后到期
+ IotScriptSource pending = IotScriptSource.of(2L, "function decode(ctx) { return {method: 'thing.property.post'}; }");
+ IotScriptResult result = executor.execute(pending, IotScriptUtils.FUNCTION_DECODE, CTX,
+ System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(100));
+ // 断言：调用方按截止时间返回超时；任务仍在 JS 池的队列里，没有启动
+ assertEquals(IotScriptResult.Status.TIMEOUT, result.getStatus());
+ assertEquals(1, jsPool.getQueue().size());
+
+ // 调用：放开 JS 线程，再提交一个探测任务；单线程 FIFO，探测任务执行时，排队超时的任务已经出队
+ release.countDown();
+ Future
+ * 关闭时整体绕开脚本：不执行、不预热；raw 上行纯透传，标准下行与未开启时相同，raw.push 照常写出
+ */
+ private boolean enabled = false;
+ /**
+ * 透传处理的截止时间：raw 上行（含 HTTP / CoAP 同一请求内应答）、脚本下行、透传回复，从收到（入队）开始计算，
+ * 覆盖排队与执行脚本
+ */
+ @NotNull(message = "透传处理截止时间不能为空")
+ private Duration deadline = Duration.ofMillis(2000);
+ /**
+ * 单次脚本执行的时间预算：与截止时间取较早者
+ */
+ @NotNull(message = "脚本执行时间预算不能为空")
+ private Duration execBudget = Duration.ofMillis(500);
+ /**
+ * 单次脚本执行的指令上限
+ */
+ private long maxInstructions = 1_000_000;
+ /**
+ * 每个设备最多排队的脚本下行指令数（不含执行中的一条）：超过时丢弃并记日志
+ */
+ @Positive(message = "每个设备最多排队的下行指令数必须大于 0")
+ private int maxPendingCommandsPerDevice = 256;
+
+ }
+
+ /**
* 协议实例配置
*/
@Data
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/enums/ErrorCodeConstants.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/enums/ErrorCodeConstants.java
index 90afda2..1095c4d 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/enums/ErrorCodeConstants.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/enums/ErrorCodeConstants.java
@@ -16,4 +16,19 @@ public interface ErrorCodeConstants {
// ========== 设备信息 1-050-002-000 ============
ErrorCode DEVICE_NOT_EXISTS = new ErrorCode(1_051_002_001, "设备({}/{}) 不存在");
+ // ========== 设备透传与数据解析脚本 1-051-003-000 ============
+ ErrorCode DEVICE_MESSAGE_RAW_METHOD_RESERVED = new ErrorCode(1_051_003_000, "透传方法只能由网关 raw 入口使用");
+ ErrorCode SCRIPT_EXECUTE_FAILED = new ErrorCode(1_051_003_100, "数据解析脚本执行失败：{}");
+ ErrorCode SCRIPT_TIMEOUT = new ErrorCode(1_051_003_101, "处理超时：未在截止时间内完成（排队、加载配置或脚本执行）");
+ ErrorCode SCRIPT_REJECTED = new ErrorCode(1_051_003_102, "数据解析脚本繁忙或已熔断");
+ ErrorCode SCRIPT_OUTPUT_INVALID = new ErrorCode(1_051_003_103, "数据解析脚本输出不合法：{}");
+ ErrorCode RAW_NOT_SUPPORTED = new ErrorCode(1_051_003_105, "当前连接不支持透传下发");
+ ErrorCode DEVICE_OFFLINE_OR_WRITE_FAILED = new ErrorCode(1_051_003_106, "设备不在线或写入失败");
+ ErrorCode MODBUS_TIMEOUT = new ErrorCode(1_051_003_110, "Modbus 响应超时");
+ ErrorCode MODBUS_EXCEPTION = new ErrorCode(1_051_003_111, "Modbus 异常响应（异常码 {}）");
+ ErrorCode MODBUS_PDU_INVALID = new ErrorCode(1_051_003_112, "Modbus PDU 不合法：{}");
+ ErrorCode MODBUS_QUEUE_FULL = new ErrorCode(1_051_003_113, "Modbus 事务队列已满");
+ ErrorCode MODBUS_RESPONSE_INVALID = new ErrorCode(1_051_003_114, "Modbus 响应不合法");
+ ErrorCode MODBUS_DISCONNECTED = new ErrorCode(1_051_003_115, "Modbus 连接已断开");
+
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java
index efd61e1..3b685e2 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java
@@ -1,13 +1,31 @@
package cn.iocoder.yudao.module.iot.gateway.protocol;
import cn.hutool.core.util.StrUtil;
+import cn.hutool.extra.spring.SpringUtil;
+import cn.iocoder.yudao.framework.common.exception.ErrorCode;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptSource;
import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
+import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
+import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.gateway.service.device.message.dto.IotDownstreamEncodeRespDTO;
+import io.vertx.core.Future;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import java.util.ArrayDeque;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+
+import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
+import static cn.iocoder.yudao.module.iot.gateway.enums.ErrorCodeConstants.RAW_NOT_SUPPORTED;
+
/**
* IoT 协议下行消息订阅者抽象类
*
@@ -23,6 +41,34 @@ public abstract class AbstractIotProtocolDownstreamSubscriber implements IotMess
private final IotMessageBus messageBus;
+ /**
+ * 有排队 / 执行中脚本指令的设备：设备编号 → 待执行的指令（第一条已在 worker 上执行时，队列可为空）
+ *
+ * 总线的消费线程与 worker 都会访问，用它自身加锁；设备的指令全部执行完后移除
+ */
+ private final Map
+ * 失败 = ServiceException(ErrorCode)，只记日志；协议自己的字节规则（例如请求 PDU）在写出之前校验，不合法返回失败。
+ * 成功：一般为本地异步写出已完成（MQTT 为 publish 已提交，晚到的写出失败由协议记日志），不代表设备已确认；有请求-响应事务的协议，提交事务之前的失败返回失败 Future，
+ * 提交之后 Future 在事务终态（响应 / 超时 / 断线 / 写失败）成功完成，网关判定的失败只记日志
+ *
+ * @param message 下行消息（raw.push，或 encode 的 property.set / service.invoke）
+ * @param bytes 要写出的字节
+ * @param source encode 时使用的脚本（固定 hash）；raw.push 为 null
+ * @return 写出结果
+ */
+ public Future
+ * 只用于设备的实际响应（包括协议的异常响应）；网关判定的失败（超时、断线、写失败等）由协议只记日志，
+ * 不调用本方法，也不给 biz 构造回复
+ *
+ * @param request 原请求
+ * @param source encode 时固定的脚本；raw.push 为 null
+ * @param payload 设备的响应字节
+ * @param deviceError 协议判定的设备错误（例如异常响应）；正常响应为 null
+ */
+ protected final void replyRawResult(IotDeviceMessage request, IotScriptSource source,
+ byte[] payload, ErrorCode deviceError) {
+ String serverId = getServerId();
+ protocol.getVertx().executeBlocking(() -> {
+ SpringUtil.getBean(IotDeviceMessageService.class).sendRawReply(request, source, payload, deviceError, serverId);
+ return null;
+ }, false).onFailure(e -> log.error("[replyRawResult][下行({}) 回复失败]", request.getRequestId(), e));
+ }
+
}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocol.java
index bdfc28b..6f458ac 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocol.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocol.java
@@ -1,6 +1,7 @@
package cn.iocoder.yudao.module.iot.gateway.protocol;
import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
+import io.vertx.core.Vertx;
/**
* IoT 协议接口
@@ -49,4 +50,13 @@ public interface IotProtocol {
*/
boolean isRunning();
+ /**
+ * 获取协议的 Vert.x 实例：阻塞的处理（数据解析脚本、发送消息）切到它的 worker 上执行
+ *
+ * @return Vert.x 实例；没有使用 Vert.x 的协议（CoAP）返回 null
+ */
+ default Vertx getVertx() {
+ return null;
+ }
+
}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocol.java
index 4bc8cdb..2425910 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocol.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/IotCoapProtocol.java
@@ -62,6 +62,11 @@ public class IotCoapProtocol implements IotProtocol {
private CoapServer coapServer;
/**
+ * 上行处理器：停止时关闭它的透传截止计时器
+ */
+ private IotCoapUpstreamHandler upstreamHandler;
+
+ /**
* 下行消息订阅者
*/
private IotCoapDownstreamSubscriber downstreamSubscriber;
@@ -116,7 +121,7 @@ public class IotCoapProtocol implements IotProtocol {
add(registerSubResource);
}});
// 2.4 添加 /topic 根资源（用于上行消息）
- IotCoapUpstreamHandler upstreamHandler = new IotCoapUpstreamHandler(serverId);
+ upstreamHandler = new IotCoapUpstreamHandler(serverId);
IotCoapUpstreamTopicResource topicResource = new IotCoapUpstreamTopicResource(serverId, upstreamHandler);
coapServer.add(topicResource);
@@ -168,6 +173,11 @@ public class IotCoapProtocol implements IotProtocol {
log.error("[stop][IoT CoAP 协议 {} 服务器停止失败]", getId(), e);
}
}
+ // 3. 关闭上行处理器的透传截止计时器
+ if (upstreamHandler != null) {
+ upstreamHandler.stop();
+ upstreamHandler = null;
+ }
running = false;
log.info("[stop][IoT CoAP 协议 {} 已停止]", getId());
}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/upstream/IotCoapAbstractHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/upstream/IotCoapAbstractHandler.java
index 994fb14..a024354 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/upstream/IotCoapAbstractHandler.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/upstream/IotCoapAbstractHandler.java
@@ -36,6 +36,11 @@ public abstract class IotCoapAbstractHandler {
*/
public static final int OPTION_TOKEN = 2088;
+ /**
+ * handle0 返回这个哨兵：表示响应由子类异步写出（透传上报：worker 处理结果与截止计时只响应一次），基类不再写响应
+ */
+ protected static final CommonResult
+ *
+ */
+package cn.iocoder.yudao.module.iot.core.script;
diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotRhinoScriptExecutor.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotRhinoScriptExecutor.java
new file mode 100644
index 0000000..903e6eb
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/script/rhino/IotRhinoScriptExecutor.java
@@ -0,0 +1,505 @@
+package cn.iocoder.yudao.module.iot.core.script.rhino;
+
+import cn.hutool.core.thread.NamedThreadFactory;
+import cn.hutool.core.util.StrUtil;
+import cn.iocoder.yudao.module.iot.core.script.IotScriptException;
+import cn.iocoder.yudao.module.iot.core.script.IotScriptExecutor;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptResult;
+import cn.iocoder.yudao.module.iot.core.script.dto.IotScriptSource;
+import lombok.Data;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+// DONE @AI：字段注释；
+/**
+ * IoT 数据解析脚本执行器：同一个 JVM 内执行，不是沙箱
+ *
+ *
+ * 不跨线程共享 Script，也不跨线程共享作用域
+ *
+ * @author 芋道源码
+ */
+final class IotScriptRuntime {
+
+ /**
+ * 脚本输出（序列化后的 JSON）的最大字符数，超过按输出不合法处理
+ */
+ static final int MAX_RESULT_CHARS = 65536;
+ /**
+ * 编译缓存最多保留的脚本数：超过时淘汰最久未使用的
+ */
+ private static final int MAX_COMPILED_SCRIPTS = 64;
+
+ /**
+ * IoT 专用的 Rhino ContextFactory：每次编译、调用都由它进入 Context
+ */
+ private final IotScriptContextFactory factory;
+ /**
+ * 创建本运行时的 JS 线程：只能在该线程上使用
+ */
+ private final Thread owner = Thread.currentThread();
+ /**
+ * 封存的标准对象：每次调用的作用域都以它为原型（prototype），作用域自身没有父作用域
+ */
+ private final ScriptableObject sealedStandard;
+ /**
+ * 封存的标准对象里的 JSON 对象：宿主只用它做 parse/stringify，用户脚本里的同名变量影响不到
+ */
+ private final Scriptable json;
+ /**
+ * 封存的 JSON.parse：把平台构造的入参 JSON 转为脚本里的 ctx 对象
+ */
+ private final Function jsonParse;
+ /**
+ * 封存的 JSON.stringify：把脚本的返回值序列化为结果 JSON
+ */
+ private final Function jsonStringify;
+ /**
+ * 编译缓存：key = productId:sha256（调试时为 debug:sha256）；只在本线程访问
+ */
+ private final Map