szz
/
ShiShiYiBan
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

【功能完善】IoT: 更新设备数据 API，重构保存设备数据方法以使用 DTO，新增参数校验依赖，优化插件管理功能，添加插件实例上报和状态更新接口，同时更新插件信息获取逻辑，删除不再使用的文件和配置。

This commit is contained in:
安浩浩 2025-01-07 17:44:55 +08:00
parent b5856c4cfc
commit cde6ebf921
21 changed files with 362 additions and 245 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -51,3 +51,4 @@ rebel.xml
application-my.yaml
/yudao-ui-app/unpackage/
**/.DS_Store

5
.vscode/settings.json vendored Normal file
View File

@ -0,0 +1,5 @@
{
"java.compile.nullAnalysis.mode": "automatic",
"java.jdt.ls.vmargs": "-XX:+UseParallelGC -XX:GCTimeRatio=4 -XX:AdaptiveSizePolicyWeight=90 -Dsun.zip.disableMemoryMapping=true -Xmx2G -Xms100m -Xlog:disable",
"java.configuration.updateBuildConfiguration": "interactive"
}

0
plugins/disabled.txt
View File

1
plugins/enabled.txt
View File

@ -1 +0,0 @@
http-plugin

BIN
plugins/yudao-module-iot-http-plugin-2.2.0-snapshot.jar
View File

Binary file not shown.

7
yudao-module-iot/yudao-module-iot-api/pom.xml
View File

@ -33,6 +33,13 @@
</exclusion>
</exclusions>
</dependency>
<!-- 参数校验 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
<optional>true</optional>
</dependency>
</dependencies>
</project>

10
yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/api/device/DeviceDataApi.java
View File

@ -1,5 +1,8 @@
package cn.iocoder.yudao.module.iot.api.device;
import cn.iocoder.yudao.module.iot.api.device.dto.DeviceDataCreateReqDTO;
import jakarta.validation.Valid;
/**
* 设备数据 API
*
@ -7,14 +10,11 @@ package cn.iocoder.yudao.module.iot.api.device;
*/
public interface DeviceDataApi {
// TODO @haohao：最好搞成 dto 哈！
/**
* 保存设备数据
*
* @param productKey 产品 key
* @param deviceName 设备名称
* @param message 消息
* @param createDTO 设备数据
*/
void saveDeviceData(String productKey, String deviceName, String message);
void saveDeviceData(@Valid DeviceDataCreateReqDTO createDTO);
}

31
yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/api/device/dto/DeviceDataCreateReqDTO.java Normal file
View File

@ -0,0 +1,31 @@
package cn.iocoder.yudao.module.iot.api.device.dto;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import jakarta.validation.constraints.NotNull;
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class DeviceDataCreateReqDTO {
/**
* 产品标识
*/
@NotNull(message = "产品标识不能为空")
private String productKey;
/**
* 设备名称
*/
@NotNull(message = "设备名称不能为空")
private String deviceName;
/**
* 消息
*/
@NotNull(message = "消息不能为空")
private String message;
}

4
yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/plugin/IotPluginDeployTypeEnum.java
View File

@ -13,8 +13,8 @@ import java.util.Arrays;
@Getter
public enum IotPluginDeployTypeEnum implements IntArrayValuable {
UPLOAD(0, "上传 jar"), // TODO @haohao：UPLOAD 和 ALONE 感觉有点冲突，前者是部署方式，后者是运行方式。这个后续再讨论下哈
ALONE(1, "独立运行");
DEPLOY_VIA_JAR(0, "通过 jar 部署"), // TODO @haohao：UPLOAD 和 ALONE 感觉有点冲突，前者是部署方式，后者是运行方式。这个后续再讨论下哈
DEPLOY_STANDALONE(1, "独立部署");
public static final int[] ARRAYS = Arrays.stream(values()).mapToInt(IotPluginDeployTypeEnum::getDeployType).toArray();

5
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/DeviceDataApiImpl.java
View File

@ -1,5 +1,6 @@
package cn.iocoder.yudao.module.iot.api.device;
import cn.iocoder.yudao.module.iot.api.device.dto.DeviceDataCreateReqDTO;
import cn.iocoder.yudao.module.iot.service.device.IotDevicePropertyDataService;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
@ -17,8 +18,8 @@ public class DeviceDataApiImpl implements DeviceDataApi {
private IotDevicePropertyDataService deviceDataService;
@Override
public void saveDeviceData(String productKey, String deviceName, String message) {
deviceDataService.saveDeviceData(productKey, deviceName, message);
public void saveDeviceData(DeviceDataCreateReqDTO createDTO) {
deviceDataService.saveDeviceData(createDTO);
}
}

5
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/plugin/vo/PluginInfoSaveReqVO.java
View File

@ -1,7 +1,9 @@
package cn.iocoder.yudao.module.iot.controller.admin.plugin.vo;
import cn.iocoder.yudao.framework.common.validation.InEnum;
import cn.iocoder.yudao.module.iot.enums.plugin.IotPluginStatusEnum;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.*;
import lombok.Data;
@Schema(description = "管理后台 - IoT 插件信息新增/修改 Request VO")
@Data
@ -39,6 +41,7 @@ public class PluginInfoSaveReqVO {
private String protocol;
@Schema(description = "状态", requiredMode = Schema.RequiredMode.REQUIRED)
@InEnum(IotPluginStatusEnum.class)
private Integer status;
@Schema(description = "插件配置项描述信息")

11
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/emq/service/EmqxServiceImpl.java
View File

@ -1,12 +1,12 @@
package cn.iocoder.yudao.module.iot.emq.service;
import cn.iocoder.yudao.module.iot.api.device.dto.DeviceDataCreateReqDTO;
import cn.iocoder.yudao.module.iot.service.device.IotDevicePropertyDataService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
// TODO @芋艿：在瞅瞅
@ -16,7 +16,7 @@ import org.springframework.stereotype.Service;
* @author ahh
*/
@Slf4j
//@Service
// @Service
public class EmqxServiceImpl implements EmqxService {
@Resource
@ -34,7 +34,12 @@ public class EmqxServiceImpl implements EmqxService {
String productKey = topic.split("/")[2];
String deviceName = topic.split("/")[3];
String message = new String(mqttMessage.getPayload());
iotDeviceDataService.saveDeviceData(productKey, deviceName, message);
DeviceDataCreateReqDTO createDTO = DeviceDataCreateReqDTO.builder()
.productKey(productKey)
.deviceName(deviceName)
.message(message)
.build();
iotDeviceDataService.saveDeviceData(createDTO);
}
}

2
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/plugin/PluginInstancesJob.java
View File

@ -22,7 +22,7 @@ public class PluginInstancesJob {
@Scheduled(initialDelay = 60, fixedRate = 60, timeUnit = TimeUnit.SECONDS)
public void updatePluginInstances() {
TenantUtils.executeIgnore(() -> {
pluginInstanceService.updatePluginInstances();
pluginInstanceService.reportPluginInstances();
});
}

8
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDevicePropertyDataService.java
View File

@ -1,6 +1,7 @@
package cn.iocoder.yudao.module.iot.service.device;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.module.iot.api.device.dto.DeviceDataCreateReqDTO;
import cn.iocoder.yudao.module.iot.controller.admin.device.vo.deviceData.IotDeviceDataPageReqVO;
import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDataDO;
import jakarta.validation.Valid;
@ -25,12 +26,9 @@ public interface IotDevicePropertyDataService {
/**
* 保存设备数据
*
* @param productKey 产品 key
* @param deviceName 设备名称
* @param message 消息
* <p>参见 <a href="https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services?spm=a2c4g.11186623.0.0.3a3335aeUdzkz2#concept-mvc-4tw-y2b">JSON 格式</a>
* @param createDTO 设备数据
*/
void saveDeviceData(String productKey, String deviceName, String message);
void saveDeviceData(DeviceDataCreateReqDTO createDTO);
/**
* 获得设备属性最新数据

17
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDevicePropertyDataServiceImpl.java
View File

@ -6,6 +6,7 @@ import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.module.iot.api.device.dto.DeviceDataCreateReqDTO;
import cn.iocoder.yudao.module.iot.controller.admin.device.vo.deviceData.IotDeviceDataPageReqVO;
import cn.iocoder.yudao.module.iot.controller.admin.thingmodel.model.dataType.ThingModelDateOrTextDataSpecs;
import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
@ -14,8 +15,8 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
import cn.iocoder.yudao.module.iot.dal.dataobject.tdengine.SelectVisualDO;
import cn.iocoder.yudao.module.iot.dal.dataobject.tdengine.ThingModelMessage;
import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
import cn.iocoder.yudao.module.iot.dal.tdengine.IotDevicePropertyDataMapper;
import cn.iocoder.yudao.module.iot.dal.redis.deviceData.DeviceDataRedisDAO;
import cn.iocoder.yudao.module.iot.dal.tdengine.IotDevicePropertyDataMapper;
import cn.iocoder.yudao.module.iot.dal.tdengine.TdEngineDMLMapper;
import cn.iocoder.yudao.module.iot.enums.IotConstants;
import cn.iocoder.yudao.module.iot.enums.thingmodel.IotDataSpecsDataTypeEnum;
@ -56,7 +57,7 @@ public class IotDevicePropertyDataServiceImpl implements IotDevicePropertyDataSe
.put(IotDataSpecsDataTypeEnum.FLOAT.getDataType(), TDengineTableField.TYPE_FLOAT)
.put(IotDataSpecsDataTypeEnum.DOUBLE.getDataType(), TDengineTableField.TYPE_DOUBLE)
.put(IotDataSpecsDataTypeEnum.ENUM.getDataType(), TDengineTableField.TYPE_TINYINT) // TODO 芋艿：为什么要映射为 TINYINT 的说明？
.put( IotDataSpecsDataTypeEnum.BOOL.getDataType(), TDengineTableField.TYPE_TINYINT) // TODO 芋艿：为什么要映射为 TINYINT 的说明？
.put(IotDataSpecsDataTypeEnum.BOOL.getDataType(), TDengineTableField.TYPE_TINYINT) // TODO 芋艿：为什么要映射为 TINYINT 的说明？
.put(IotDataSpecsDataTypeEnum.TEXT.getDataType(), TDengineTableField.TYPE_NCHAR)
.put(IotDataSpecsDataTypeEnum.DATE.getDataType(), TDengineTableField.TYPE_TIMESTAMP)
.put(IotDataSpecsDataTypeEnum.STRUCT.getDataType(), TDengineTableField.TYPE_NCHAR) // TODO 芋艿：怎么映射！！！！
@ -128,20 +129,20 @@ public class IotDevicePropertyDataServiceImpl implements IotDevicePropertyDataSe
}
@Override
public void saveDeviceData(String productKey, String deviceName, String message) {
public void saveDeviceData(DeviceDataCreateReqDTO createDTO) {
// 1. 根据产品 key 和设备名称，获得设备信息
IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceName(productKey, deviceName);
IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceName(createDTO.getProductKey(), createDTO.getDeviceName());
// 2. 解析消息，保存数据
JSONObject jsonObject = new JSONObject(message);
log.info("[saveDeviceData][productKey({}) deviceName({}) data({})]", productKey, deviceName, jsonObject);
JSONObject jsonObject = new JSONObject(createDTO.getMessage());
log.info("[saveDeviceData][productKey({}) deviceName({}) data({})]", createDTO.getProductKey(), createDTO.getDeviceName(), jsonObject);
ThingModelMessage thingModelMessage = ThingModelMessage.builder()
.id(jsonObject.getStr("id"))
.sys(jsonObject.get("sys"))
.method(jsonObject.getStr("method"))
.params(jsonObject.get("params"))
.time(jsonObject.getLong("time") == null ? System.currentTimeMillis() : jsonObject.getLong("time"))
.productKey(productKey)
.deviceName(deviceName)
.productKey(createDTO.getProductKey())
.deviceName(createDTO.getDeviceName())
.deviceKey(device.getDeviceKey())
.build();
thingModelMessageService.saveThingModelMessage(device, thingModelMessage);

7
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/PluginInfoService.java
View File

@ -78,9 +78,10 @@ public interface PluginInfoService {
List<PluginInfoDO> getPluginInfoList();
/**
* 获得运行状态的插件信息列表
* 根据状态获得插件信息列表
*
* @return 运行状态的插件信息列表
* @param status 状态
* @return 插件信息列表
*/
List<PluginInfoDO> getRunningPluginInfoList();
List<PluginInfoDO> getPluginInfoListByStatus(Integer status);
}

204
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/PluginInfoServiceImpl.java
View File

@ -9,25 +9,16 @@ import cn.iocoder.yudao.module.iot.dal.mysql.plugin.PluginInfoMapper;
import cn.iocoder.yudao.module.iot.enums.plugin.IotPluginStatusEnum;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.pf4j.PluginDescriptor;
import org.pf4j.PluginState;
import org.pf4j.PluginWrapper;
import org.pf4j.spring.SpringPluginManager;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.IOException;
import java.nio.file.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.*;
import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.PLUGIN_INFO_DELETE_FAILED_RUNNING;
import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.PLUGIN_INFO_NOT_EXISTS;
/**
* IoT 插件信息 Service 实现类
@ -43,11 +34,10 @@ public class PluginInfoServiceImpl implements PluginInfoService {
private PluginInfoMapper pluginInfoMapper;
@Resource
private SpringPluginManager pluginManager;
private PluginInstanceService pluginInstanceService;
// TODO @芋艿：要不要换位置
@Value("${pf4j.pluginsDir}")
private String pluginsDir;
@Resource
private SpringPluginManager pluginManager;
@Override
public Long createPluginInfo(PluginInfoSaveReqVO createReqVO) {
@ -75,32 +65,13 @@ public class PluginInfoServiceImpl implements PluginInfoService {
}
// 2. 卸载插件
// TODO @haohao：可以复用 stopAndUnloadPlugin
PluginWrapper plugin = pluginManager.getPlugin(pluginInfoDO.getPluginKey());
if (plugin != null) {
// 停止插件
if (plugin.getPluginState().equals(PluginState.STARTED)) {
pluginManager.stopPlugin(plugin.getPluginId());
}
// 卸载插件
pluginManager.unloadPlugin(plugin.getPluginId());
}
pluginInstanceService.stopAndUnloadPlugin(pluginInfoDO.getPluginKey());
// 3.1 删除
// 3. 删除插件文件
pluginInstanceService.deletePluginFile(pluginInfoDO);
// 4. 删除插件信息
pluginInfoMapper.deleteById(id);
// 3.2 删除插件文件
// TODO @haohao：这个直接主线程 sleep 就好了，不用单独开线程池哈。原因是，低频操作；另外，只有存在的时候，才 sleep + 删除；
Executors.newSingleThreadExecutor().submit(() -> {
try {
TimeUnit.SECONDS.sleep(1); // 等待 1 秒，避免插件未卸载完毕
File file = new File(pluginsDir, pluginInfoDO.getFileName());
if (file.exists() && !file.delete()) {
log.error("[deletePluginInfo][删除插件文件({}) 失败]", pluginInfoDO.getFileName());
}
} catch (InterruptedException e) {
log.error("[deletePluginInfo][删除插件文件({}) 失败]", pluginInfoDO.getFileName(), e);
}
});
}
private PluginInfoDO validatePluginInfoExists(Long id) {
@ -127,144 +98,52 @@ public class PluginInfoServiceImpl implements PluginInfoService {
PluginInfoDO pluginInfoDo = validatePluginInfoExists(id);
// 2. 停止并卸载旧的插件
stopAndUnloadPlugin(pluginInfoDo.getPluginKey());
pluginInstanceService.stopAndUnloadPlugin(pluginInfoDo.getPluginKey());
// 3.1 上传新的插件文件
String pluginKeyNew = uploadAndLoadNewPlugin(file);
// 3.2 更新插件启用状态文件
updatePluginStatusFile(pluginKeyNew, false);
// 3 上传新的插件文件,更新插件启用状态文件
String pluginKeyNew = pluginInstanceService.uploadAndLoadNewPlugin(file);
pluginInstanceService.updatePluginStatusFile(pluginKeyNew, false);
// 4. 更新插件信息
updatePluginInfo(pluginInfoDo, pluginKeyNew, file);
}
// TODO @haohao：注释的格式
// 停止并卸载旧的插件
private void stopAndUnloadPlugin(String pluginKey) {
PluginWrapper plugin = pluginManager.getPlugin(pluginKey);
if (plugin != null) {
if (plugin.getPluginState().equals(PluginState.STARTED)) {
pluginManager.stopPlugin(pluginKey); // 停止插件
}
pluginManager.unloadPlugin(pluginKey); // 卸载插件
}
}
// TODO @haohao：注释的格式
// 上传并加载新的插件文件
private String uploadAndLoadNewPlugin(MultipartFile file) {
// TODO @haohao：多节点，是不是要上传 s3 之类的存储器；然后定时去加载
Path pluginsPath = Paths.get(pluginsDir);
try {
// TODO @haohao：可以使用 FileUtil 简化？
if (!Files.exists(pluginsPath)) {
Files.createDirectories(pluginsPath); // 创建插件目录
}
String filename = file.getOriginalFilename();
if (filename != null) {
Path jarPath = pluginsPath.resolve(filename);
Files.copy(file.getInputStream(), jarPath, StandardCopyOption.REPLACE_EXISTING); // 保存上传的 JAR 文件
return pluginManager.loadPlugin(jarPath.toAbsolutePath()); // 加载插件
}
throw exception(PLUGIN_INSTALL_FAILED); // TODO @haohao：这么抛的话，貌似会被 catch (Exception e) {
} catch (Exception e) {
// TODO @haohao：打个 error log，方便排查
throw exception(PLUGIN_INSTALL_FAILED);
}
}
// TODO @haohao：注释的格式
// 更新插件状态文件
private void updatePluginStatusFile(String pluginKeyNew, boolean isEnabled) {
// TODO @haohao：疑问，这里写 enabled.txt 和 disabled.txt 的目的是啥哈？
Path enabledFilePath = Paths.get(pluginsDir, "enabled.txt");
Path disabledFilePath = Paths.get(pluginsDir, "disabled.txt");
Path targetFilePath = isEnabled ? enabledFilePath : disabledFilePath;
Path oppositeFilePath = isEnabled ? disabledFilePath : enabledFilePath;
try {
PluginWrapper pluginWrapper = pluginManager.getPlugin(pluginKeyNew);
if (pluginWrapper == null) {
throw exception(PLUGIN_INSTALL_FAILED);
}
List<String> targetLines = Files.exists(targetFilePath) ? Files.readAllLines(targetFilePath) : new ArrayList<>();
List<String> oppositeLines = Files.exists(oppositeFilePath) ? Files.readAllLines(oppositeFilePath) : new ArrayList<>();
if (!targetLines.contains(pluginKeyNew)) {
targetLines.add(pluginKeyNew);
Files.write(targetFilePath, targetLines, StandardOpenOption.CREATE,
StandardOpenOption.TRUNCATE_EXISTING);
}
if (oppositeLines.contains(pluginKeyNew)) {
oppositeLines.remove(pluginKeyNew);
Files.write(oppositeFilePath, oppositeLines, StandardOpenOption.CREATE,
StandardOpenOption.TRUNCATE_EXISTING);
}
} catch (IOException e) {
throw exception(PLUGIN_INSTALL_FAILED);
}
}
// TODO @haohao：注释的格式
// 更新插件信息
/**
* 更新插件信息
*
* @param pluginInfoDo 插件信息
* @param pluginKeyNew 插件标识符
* @param file 文件
*/
private void updatePluginInfo(PluginInfoDO pluginInfoDo, String pluginKeyNew, MultipartFile file) {
// TODO @haohao：更新实体的时候，最好 new 一个新的！
// TODO @haohao：可以链式调用，简化下代码；
pluginInfoDo.setPluginKey(pluginKeyNew);
pluginInfoDo.setStatus(IotPluginStatusEnum.STOPPED.getStatus());
pluginInfoDo.setFileName(file.getOriginalFilename());
pluginInfoDo.setScript("");
// 解析 pf4j 插件
PluginDescriptor pluginDescriptor = pluginManager.getPlugin(pluginKeyNew).getDescriptor();
pluginInfoDo.setConfigSchema(pluginDescriptor.getPluginDescription());
pluginInfoDo.setVersion(pluginDescriptor.getVersion());
pluginInfoDo.setDescription(pluginDescriptor.getPluginDescription());
// 创建新的插件信息对象并链式设置属性
PluginInfoDO updatedPluginInfo = new PluginInfoDO()
.setId(pluginInfoDo.getId())
.setPluginKey(pluginKeyNew)
.setStatus(IotPluginStatusEnum.STOPPED.getStatus())
.setFileName(file.getOriginalFilename())
.setScript("")
.setConfigSchema(pluginManager.getPlugin(pluginKeyNew).getDescriptor().getPluginDescription())
.setVersion(pluginManager.getPlugin(pluginKeyNew).getDescriptor().getVersion())
.setDescription(pluginManager.getPlugin(pluginKeyNew).getDescriptor().getPluginDescription());
// 执行更新
pluginInfoMapper.updateById(pluginInfoDo);
pluginInfoMapper.updateById(updatedPluginInfo);
}
// TODO @haohao：status、state 字段命名，要统一下~
@Override
public void updatePluginStatus(Long id, Integer status) {
// 1. 校验插件信息是否存在
PluginInfoDO pluginInfoDo = validatePluginInfoExists(id);
// 2. 校验插件状态是否有效
// TODO @haohao：直接参数校验掉。通过 @InEnum
if (!IotPluginStatusEnum.contains(status)) {
throw exception(PLUGIN_STATUS_INVALID);
}
// 2. 更新插件状态
pluginInstanceService.updatePluginStatus(pluginInfoDo, status);
// 3. 获取插件标识和插件实例
String pluginKey = pluginInfoDo.getPluginKey();
PluginWrapper plugin = pluginManager.getPlugin(pluginKey);
// 4. 根据状态更新插件
if (plugin != null) {
// 4.1 启动：如果目标状态是运行且插件未启动，则启动插件
if (status.equals(IotPluginStatusEnum.RUNNING.getStatus())
&& plugin.getPluginState() != PluginState.STARTED) {
pluginManager.startPlugin(pluginKey);
updatePluginStatusFile(pluginKey, true);
// 4.2 停止：如果目标状态是停止且插件已启动，则停止插件
} else if (status.equals(IotPluginStatusEnum.STOPPED.getStatus())
&& plugin.getPluginState() == PluginState.STARTED) {
pluginManager.stopPlugin(pluginKey);
updatePluginStatusFile(pluginKey, false);
}
} else {
// 5. 插件不存在且状态为停止，抛出异常
if (IotPluginStatusEnum.STOPPED.getStatus().equals(pluginInfoDo.getStatus())) {
throw exception(PLUGIN_STATUS_INVALID);
}
}
// 6. 更新数据库中的插件状态
// TODO @haohao：新建新建 pluginInfoDo 哈！
pluginInfoDo.setStatus(status);
pluginInfoMapper.updateById(pluginInfoDo);
// 3. 更新数据库中的插件状态
PluginInfoDO updatedPluginInfo = new PluginInfoDO();
updatedPluginInfo.setId(id);
updatedPluginInfo.setStatus(status);
pluginInfoMapper.updateById(updatedPluginInfo);
}
@Override
@ -272,10 +151,9 @@ public class PluginInfoServiceImpl implements PluginInfoService {
return pluginInfoMapper.selectList();
}
// TODO @haohao：可以改成 getPluginInfoListByStatus 更通用哈。
@Override
public List<PluginInfoDO> getRunningPluginInfoList() {
return pluginInfoMapper.selectListByStatus(IotPluginStatusEnum.RUNNING.getStatus());
public List<PluginInfoDO> getPluginInfoListByStatus(Integer status) {
return pluginInfoMapper.selectListByStatus(status);
}
}

45
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/PluginInstanceService.java
View File

@ -1,5 +1,8 @@
package cn.iocoder.yudao.module.iot.service.plugin;
import cn.iocoder.yudao.module.iot.dal.dataobject.plugininfo.PluginInfoDO;
import org.springframework.web.multipart.MultipartFile;
/**
* IoT 插件实例 Service 接口
*
@ -8,8 +11,46 @@ package cn.iocoder.yudao.module.iot.service.plugin;
public interface PluginInstanceService {
/**
* 更新IoT 插件实例
* 上报插件实例
*/
void updatePluginInstances();
void reportPluginInstances();
/**
* 停止并卸载插件
*
* @param pluginKey 插件标识符
*/
void stopAndUnloadPlugin(String pluginKey);
/**
* 删除插件文件
*
* @param pluginInfoDo 插件信息
*/
void deletePluginFile(PluginInfoDO pluginInfoDo);
/**
* 上传并加载新的插件文件
*
* @param file 插件文件
* @return 插件标识符
*/
String uploadAndLoadNewPlugin(MultipartFile file);
/**
* 更新插件状态文件
*
* @param pluginKeyNew 插件标识符
* @param isEnabled 是否启用
*/
void updatePluginStatusFile(String pluginKeyNew, boolean isEnabled);
/**
* 更新插件状态
*
* @param pluginInfoDo 插件信息
* @param status 新状态
*/
void updatePluginStatus(PluginInfoDO pluginInfoDo, Integer status);
}

227
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/PluginInstanceServiceImpl.java
View File

@ -1,19 +1,38 @@
package cn.iocoder.yudao.module.iot.service.plugin;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.IdUtil;
import cn.iocoder.yudao.module.iot.dal.dataobject.plugininfo.PluginInfoDO;
import cn.iocoder.yudao.module.iot.dal.dataobject.plugininstance.PluginInstanceDO;
import cn.iocoder.yudao.module.iot.dal.mysql.plugin.PluginInfoMapper;
import cn.iocoder.yudao.module.iot.dal.mysql.plugin.PluginInstanceMapper;
import cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants;
import cn.iocoder.yudao.module.iot.enums.plugin.IotPluginStatusEnum;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.pf4j.PluginState;
import org.pf4j.PluginWrapper;
import org.pf4j.spring.SpringPluginManager;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.IOException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.nio.file.*;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
/**
* IoT 插件实例 Service 实现类
@ -25,79 +44,195 @@ import java.util.List;
@Slf4j
public class PluginInstanceServiceImpl implements PluginInstanceService {
/**
* 主程序 ID
*/
// TODO @haohao：这个可以后续确认下，有没更合适的标识。例如说 mac 地址之类的
// 简化的UUID + mac 地址 会不会好一些，一台机子有可能会部署多个插件
public static final String MAIN_ID = IdUtil.fastSimpleUUID();
@Resource
private PluginInfoService pluginInfoService;
private PluginInfoMapper pluginInfoMapper;
@Resource
private PluginInstanceMapper pluginInstanceMapper;
@Resource
private SpringPluginManager pluginManager;
@Value("${pf4j.pluginsDir}")
private String pluginsDir;
@Value("${server.port:48080}")
private int port;
// TODO @haohao：建议把 PluginInfoServiceImpl 里面，和 instance 相关的逻辑拿过来，可能会更好。info 处理信息，instance 处理实例
// TODO @haohao：这个改成 reportPluginInstance 会不会更合适哈。
@Override
public void updatePluginInstances() {
// 1.1 查询 pf4j 插件列表
List<PluginWrapper> plugins = pluginManager.getPlugins();
// 1.2 查询插件信息列表
List<PluginInfoDO> pluginInfos = pluginInfoService.getPluginInfoList();
// 1.3 动态获取主程序的 IP 和端口
String mainIp = getLocalIpAddress();
public void stopAndUnloadPlugin(String pluginKey) {
PluginWrapper plugin = pluginManager.getPlugin(pluginKey);
if (plugin != null) {
if (plugin.getPluginState().equals(PluginState.STARTED)) {
pluginManager.stopPlugin(pluginKey); // 停止插件
log.info("已停止插件: {}", pluginKey);
}
pluginManager.unloadPlugin(pluginKey); // 卸载插件
log.info("已卸载插件: {}", pluginKey);
} else {
log.warn("插件不存在或已卸载: {}", pluginKey);
}
}
// 2. 遍历插件列表，并保存为插件实例
@Override
public void deletePluginFile(PluginInfoDO pluginInfoDO) {
File file = new File(pluginsDir, pluginInfoDO.getFileName());
if (file.exists()) {
try {
TimeUnit.SECONDS.sleep(1); // 等待 1 秒，避免插件未卸载完毕
if (!file.delete()) {
log.error("[deletePluginInfo][删除插件文件({}) 失败]", pluginInfoDO.getFileName());
}
} catch (InterruptedException e) {
log.error("[deletePluginInfo][删除插件文件({}) 失败]", pluginInfoDO.getFileName(),
e);
Thread.currentThread().interrupt(); // 恢复中断状态
}
}
}
@Override
public String uploadAndLoadNewPlugin(MultipartFile file) {
String pluginKeyNew;
// TODO @haohao：多节点，是不是要上传 s3 之类的存储器；然后定时去加载
Path pluginsPath = Paths.get(pluginsDir);
try {
FileUtil.mkdir(pluginsPath.toFile()); // 创建插件目录
String filename = file.getOriginalFilename();
if (filename != null) {
Path jarPath = pluginsPath.resolve(filename);
Files.copy(file.getInputStream(), jarPath, StandardCopyOption.REPLACE_EXISTING); // 保存上传的 JAR 文件
pluginKeyNew = pluginManager.loadPlugin(jarPath.toAbsolutePath()); // 加载插件
log.info("已加载插件: {}", pluginKeyNew);
} else {
throw exception(ErrorCodeConstants.PLUGIN_INSTALL_FAILED);
}
} catch (IOException e) {
log.error("[uploadAndLoadNewPlugin][上传插件文件失败]", e);
throw exception(ErrorCodeConstants.PLUGIN_INSTALL_FAILED, e);
} catch (Exception e) {
log.error("[uploadAndLoadNewPlugin][加载插件失败]", e);
throw exception(ErrorCodeConstants.PLUGIN_INSTALL_FAILED, e);
}
return pluginKeyNew;
}
@Override
public void updatePluginStatusFile(String pluginKeyNew, boolean isEnabled) {
// TODO @haohao：疑问，这里写 enabled.txt 和 disabled.txt 的目的是啥哈？
// pf4j 的插件状态文件，需要 2 个文件，一个 enabled.txt 一个 disabled.txt
Path enabledFilePath = Paths.get(pluginsDir, "enabled.txt");
Path disabledFilePath = Paths.get(pluginsDir, "disabled.txt");
Path targetFilePath = isEnabled ? enabledFilePath : disabledFilePath;
Path oppositeFilePath = isEnabled ? disabledFilePath : enabledFilePath;
try {
PluginWrapper pluginWrapper = pluginManager.getPlugin(pluginKeyNew);
if (pluginWrapper == null) {
throw exception(ErrorCodeConstants.PLUGIN_INSTALL_FAILED);
}
List<String> targetLines = Files.exists(targetFilePath) ? Files.readAllLines(targetFilePath)
: new ArrayList<>();
List<String> oppositeLines = Files.exists(oppositeFilePath) ? Files.readAllLines(oppositeFilePath)
: new ArrayList<>();
if (!targetLines.contains(pluginKeyNew)) {
targetLines.add(pluginKeyNew);
Files.write(targetFilePath, targetLines, StandardOpenOption.CREATE,
StandardOpenOption.TRUNCATE_EXISTING);
log.info("已添加插件 {} 到 {}", pluginKeyNew, targetFilePath.getFileName());
}
if (oppositeLines.contains(pluginKeyNew)) {
oppositeLines.remove(pluginKeyNew);
Files.write(oppositeFilePath, oppositeLines, StandardOpenOption.CREATE,
StandardOpenOption.TRUNCATE_EXISTING);
log.info("已从 {} 移除插件 {}", oppositeFilePath.getFileName(), pluginKeyNew);
}
} catch (IOException e) {
log.error("[updatePluginStatusFile][更新插件状态文件失败]", e);
throw exception(ErrorCodeConstants.PLUGIN_INSTALL_FAILED, e);
}
}
@Override
public void updatePluginStatus(PluginInfoDO pluginInfoDo, Integer status) {
String pluginKey = pluginInfoDo.getPluginKey();
PluginWrapper plugin = pluginManager.getPlugin(pluginKey);
if (plugin != null) {
// 启动插件
if (status.equals(IotPluginStatusEnum.RUNNING.getStatus())
&& plugin.getPluginState() != PluginState.STARTED) {
pluginManager.startPlugin(pluginKey);
updatePluginStatusFile(pluginKey, true);
log.info("已启动插件: {}", pluginKey);
}
// 停止插件
else if (status.equals(IotPluginStatusEnum.STOPPED.getStatus())
&& plugin.getPluginState() == PluginState.STARTED) {
pluginManager.stopPlugin(pluginKey);
updatePluginStatusFile(pluginKey, false);
log.info("已停止插件: {}", pluginKey);
}
} else {
// 插件不存在且状态为停止，抛出异常
if (IotPluginStatusEnum.STOPPED.getStatus().equals(pluginInfoDo.getStatus())) {
throw exception(ErrorCodeConstants.PLUGIN_STATUS_INVALID);
}
}
}
@Override
public void reportPluginInstances() {
// 1. 获取 pf4j 插件列表
List<PluginWrapper> plugins = pluginManager.getPlugins();
// 2. 获取插件信息列表并转换为 Map 以便快速查找
List<PluginInfoDO> pluginInfos = pluginInfoMapper.selectList();
Map<String, PluginInfoDO> pluginInfoMap = pluginInfos.stream()
.collect(Collectors.toMap(PluginInfoDO::getPluginKey, Function.identity()));
// 3. 获取本机 IP 和 MAC 地址
LinkedHashSet<InetAddress> localAddressList = NetUtil.localAddressList(t -> t instanceof Inet4Address);
LinkedHashSet<String> ipList = NetUtil.toIpList(localAddressList);
String ip = ipList.stream().findFirst().orElse("127.0.0.1");
String mac = NetUtil.getMacAddress(localAddressList.stream().findFirst().orElse(null));
String mainId = MAIN_ID + "-" + mac;
// 4. 遍历插件列表，并保存为插件实例
for (PluginWrapper plugin : plugins) {
// 2.1 查找插件信息
String pluginKey = plugin.getPluginId();
// TODO @haohao：CollUtil.findOne() 简化
PluginInfoDO pluginInfo = pluginInfos.stream()
.filter(pluginInfoDO -> pluginInfoDO.getPluginKey().equals(pluginKey))
.findFirst()
.orElse(null);
// 4.1 查找插件信息
PluginInfoDO pluginInfo = pluginInfoMap.get(pluginKey);
if (pluginInfo == null) {
// TODO @haohao：建议打个 error log
// 4.2 插件信息不存在，记录错误并跳过
log.error("插件信息不存在，插件包标识符 = {}", pluginKey);
continue;
}
// 2.2 查询插件实例
PluginInstanceDO pluginInstance = pluginInstanceMapper.selectByMainIdAndPluginId(MAIN_ID, pluginInfo.getId());
// 2.3.1 如果插件实例不存在，则创建
// 4.3 查询插件实例
PluginInstanceDO pluginInstance = pluginInstanceMapper.selectByMainIdAndPluginId(mainId,
pluginInfo.getId());
if (pluginInstance == null) {
// TODO @haohao：可以链式调用；建议新建一个！
pluginInstance = new PluginInstanceDO();
pluginInstance.setPluginId(pluginInfo.getId());
pluginInstance.setMainId(MAIN_ID);
pluginInstance.setIp(mainIp);
pluginInstance.setPort(port);
pluginInstance.setHeartbeatAt(System.currentTimeMillis());
// 4.4 如果插件实例不存在，则创建
pluginInstance = PluginInstanceDO.builder()
.pluginId(pluginInfo.getId())
.mainId(MAIN_ID + "-" + mac)
.ip(ip)
.port(port)
.heartbeatAt(System.currentTimeMillis())
.build();
pluginInstanceMapper.insert(pluginInstance);
} else {
// 2.3.2 如果插件实例存在，则更新
// 4.5 如果插件实例存在，则更新心跳时间
pluginInstance.setHeartbeatAt(System.currentTimeMillis());
pluginInstanceMapper.updateById(pluginInstance);
}
}
}
// TODO @haohao：这个目的是，获取到第一个有效 ip 是哇？
private String getLocalIpAddress() {
try {
List<String> ipList = NetUtil.localIpv4s().stream()
.filter(ip -> !ip.startsWith("0.0") && !ip.startsWith("127.") && !ip.startsWith("169.254") && !ip.startsWith("255.255.255.255"))
.toList();
return ipList.isEmpty() ? "127.0.0.1" : ipList.get(0);
} catch (Exception e) {
log.error("获取本地IP地址失败", e);
return "127.0.0.1"; // 默认值
}
}
}

5
yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/controller/RpcController.java
View File

@ -11,6 +11,11 @@ import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.concurrent.CompletableFuture;
/**
* 插件实例 RPC 接口
*
* @author 芋道源码
*/
@RestController
@RequestMapping("/rpc")
@RequiredArgsConstructor

10
yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/plugin/HttpHandler.java
View File

@ -3,6 +3,7 @@ package cn.iocoder.yudao.module.iot.plugin;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.module.iot.api.device.DeviceDataApi;
import cn.iocoder.yudao.module.iot.api.device.dto.DeviceDataCreateReqDTO;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
@ -12,7 +13,7 @@ import io.netty.util.CharsetUtil;
/**
* 基于 Netty 的 HTTP 处理器，用于接收设备上报的数据并调用主程序的 DeviceDataApi 接口进行处理。
*
* <p>
* 1. 请求格式：JSON 格式，地址为 POST /sys/{productKey}/{deviceName}/thing/event/property/post
* 2. 返回结果：JSON 格式，包含统一的 code、data、id、message、method、version 字段
*/
@ -76,7 +77,12 @@ public class HttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
try {
// 调用主程序的接口保存数据
deviceDataApi.saveDeviceData(productKey, deviceName, jsonData.toString());
DeviceDataCreateReqDTO createDTO = DeviceDataCreateReqDTO.builder()
.productKey(productKey)
.deviceName(deviceName)
.message(jsonData.toString())
.build();
deviceDataApi.saveDeviceData(createDTO);
// 构造成功响应内容
JSONObject successRes = createResponseJson(