diff --git a/fastbee-admin/src/main/resources/application.yml b/fastbee-admin/src/main/resources/application.yml index 1790faef..c5ad1275 100644 --- a/fastbee-admin/src/main/resources/application.yml +++ b/fastbee-admin/src/main/resources/application.yml @@ -219,6 +219,12 @@ liteflow: scriptEnableField: enable +# 离线待下发指令（设备上线后自动补发 FUNCTION_GET） +iot: + deferred-command: + expire-hours: 72 + flush-delay-ms: 1000 + # 地图配置 map: provider: gaode # 提供者 diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/DeviceMessage.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/DeviceMessage.java index ef426912..51892f06 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/DeviceMessage.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/DeviceMessage.java @@ -32,6 +32,11 @@ public class DeviceMessage { private String dataType; + /** + * 为 true 且设备离线时，将 FUNCTION_GET 写入待执行队列，等设备上线后再下发 + */ + private Boolean deferIfOffline; + /** * 时间 */ diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/DeviceMessagePostResult.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/DeviceMessagePostResult.java new file mode 100644 index 00000000..c6748bcf --- /dev/null +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/DeviceMessagePostResult.java @@ -0,0 +1,32 @@ +package com.fastbee.common.core.mq.message; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * 平台下发指令结果（支持离线入队） + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class DeviceMessagePostResult { + + /** + * 是否因离线已写入待执行队列 + */ + private boolean deferred; + + /** + * 待执行任务 id，deferred 为 true 时有值 + */ + private Long taskId; + + public static DeviceMessagePostResult sent() { + return new DeviceMessagePostResult(false, null); + } + + public static DeviceMessagePostResult deferred(Long taskId) { + return new DeviceMessagePostResult(true, taskId); + } +} diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java index 631a6b71..4a330457 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DevicePropFetchConsumer.java @@ -132,7 +132,7 @@ public class DevicePropFetchConsumer { modbusJob.setStatus("0"); List modbusJobList = modbusJobService.selectModbusJobList(modbusJob); if (CollectionUtils.isEmpty(modbusJobList)) return modbusPollMsg; - //处理子设备情况 + // 处理子设备情况 handleSubDeviceStatus(modbusJobList); List commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList()); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java index 4a794907..3301ce9a 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceStatusConsumer.java @@ -10,6 +10,7 @@ import com.fastbee.common.core.mq.message.ReportDataBo; import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem; import com.fastbee.common.enums.DeviceStatus; import com.fastbee.iot.cache.IDeviceCache; +import com.fastbee.iot.data.service.IDeviceDeferredCommandService; import com.fastbee.iot.data.service.IMqttMessagePublish; import com.fastbee.iot.data.service.IRuleEngine; import com.fastbee.iot.domain.Device; @@ -57,6 +58,8 @@ public class DeviceStatusConsumer { private Boolean enabled; @Resource private ISubGatewayService subGatewayService; + @Resource + private IDeviceDeferredCommandService deviceDeferredCommandService; private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(2); public synchronized void consume(DeviceStatusBo bo) { @@ -90,9 +93,14 @@ public class DeviceStatusConsumer { } /*更新设备状态*/ device.setProtocolCode(product.getProtocolCode()); + boolean transitionToOnline = status.equals(DeviceStatus.ONLINE) + && device.getStatus() != DeviceStatus.ONLINE.getType(); deviceCache.updateDeviceStatusCache(statusBo, device); //处理影子模式值 this.handlerShadow(device, status); + if (transitionToOnline) { + deviceDeferredCommandService.flushPendingOnOnline(statusBo.getSerialNumber()); + } //设备上下线执行规则引擎 ReportDataBo dataBo = new ReportDataBo(); dataBo.setRuleEngine(true); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceDeferredCommandService.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceDeferredCommandService.java new file mode 100644 index 00000000..a91db83b --- /dev/null +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceDeferredCommandService.java @@ -0,0 +1,21 @@ +package com.fastbee.iot.data.service; + +import com.fastbee.common.core.mq.message.DeviceMessage; + +/** + * 设备离线待下发指令（上线后补发） + */ +public interface IDeviceDeferredCommandService { + + /** + * 入队 FUNCTION_GET 类指令（离线时） + * + * @return 任务主键（去重时返回已有待执行任务 id） + */ + Long enqueueFunctionGet(DeviceMessage deviceMessage); + + /** + * 设备由离线变为在线后，延迟补发待执行指令 + */ + void flushPendingOnOnline(String serialNumber); +} diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceMessageService.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceMessageService.java index 883f74a9..9f058661 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceMessageService.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/IDeviceMessageService.java @@ -1,6 +1,7 @@ package com.fastbee.iot.data.service; import com.fastbee.common.core.mq.message.DeviceMessage; +import com.fastbee.common.core.mq.message.DeviceMessagePostResult; import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem; import com.fastbee.iot.model.VariableReadVO; import com.fastbee.modbus.model.ModbusRtu; @@ -12,7 +13,15 @@ import java.util.List; */ public interface IDeviceMessageService { - void messagePost(DeviceMessage deviceMessage); + /** + * 平台下发指令（支持 deferIfOffline 离线入队） + */ + DeviceMessagePostResult messagePost(DeviceMessage deviceMessage); + + /** + * 直接下发 FUNCTION_GET（内部与上线补发使用，不再做离线入队判断） + */ + void dispatchFunctionGet(DeviceMessage deviceMessage); String messageEncode(ModbusRtu modbusRtu); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceDeferredCommandServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceDeferredCommandServiceImpl.java new file mode 100644 index 00000000..d161f417 --- /dev/null +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceDeferredCommandServiceImpl.java @@ -0,0 +1,120 @@ +package com.fastbee.iot.data.service.impl; + +import com.fastbee.common.constant.FastBeeConstant; +import com.fastbee.common.core.mq.message.DeviceMessage; +import com.fastbee.common.utils.DateUtils; +import com.fastbee.common.utils.StringUtils; +import com.fastbee.common.utils.sign.Md5Utils; +import com.fastbee.iot.data.service.IDeviceDeferredCommandService; +import com.fastbee.iot.data.service.IDeviceMessageService; +import com.fastbee.iot.domain.DeviceDeferredCommand; +import com.fastbee.iot.mapper.DeviceDeferredCommandMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Lazy; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.util.List; + +/** + * 离线待下发指令：入队与上线补发 + */ +@Slf4j +@Service +public class DeviceDeferredCommandServiceImpl implements IDeviceDeferredCommandService { + + @Resource + private DeviceDeferredCommandMapper deviceDeferredCommandMapper; + + @Lazy + @Resource + private IDeviceMessageService deviceMessageService; + + @Value("${iot.deferred-command.expire-hours:72}") + private int expireHours; + + @Value("${iot.deferred-command.flush-delay-ms:2000}") + private long flushDelayMs; + + @Override + public Long enqueueFunctionGet(DeviceMessage deviceMessage) { + String serial = deviceMessage.getSerialNumber(); + if (StringUtils.isEmpty(serial)) { + throw new IllegalArgumentException("设备编号不能为空"); + } + String topic = deviceMessage.getTopicName(); + if (StringUtils.isEmpty(topic)) { + topic = "/function/get"; + } + String messageStr = normalizeMessagePayload(deviceMessage.getMessage()); + String hash = Md5Utils.hash(serial + "|" + (topic == null ? "" : topic) + "|" + messageStr); + DeviceDeferredCommand exist = deviceDeferredCommandMapper.selectPendingBySerialAndHash(serial, hash); + if (exist != null) { + return exist.getId(); + } + DeviceDeferredCommand row = new DeviceDeferredCommand(); + row.setSerialNumber(serial); + row.setTopicName(topic); + row.setMessageHex(messageStr); + row.setDataType(deviceMessage.getDataType()); + row.setTaskType(DeviceDeferredCommand.TASK_TYPE_FUNCTION_GET); + row.setStatus(DeviceDeferredCommand.STATUS_PENDING); + row.setPayloadHash(hash); + row.setCreateTime(DateUtils.getNowDate()); + row.setExpireTime(DateUtils.addHours(row.getCreateTime(), expireHours)); + deviceDeferredCommandMapper.insert(row); + return row.getId(); + } + + @Override + @Async(FastBeeConstant.TASK.DEVICE_FETCH_PROP_TASK) + public void flushPendingOnOnline(String serialNumber) { + if (StringUtils.isEmpty(serialNumber)) { + return; + } + if (flushDelayMs > 0) { + try { + Thread.sleep(flushDelayMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + List list = deviceDeferredCommandMapper.selectPendingListBySerialNumber(serialNumber); + if (list == null || list.isEmpty()) { + return; + } + for (DeviceDeferredCommand cmd : list) { + int n = deviceDeferredCommandMapper.updateDispatchingIfPending(cmd.getId()); + if (n == 0) { + continue; + } + try { + DeviceMessage dm = new DeviceMessage(); + dm.setSerialNumber(cmd.getSerialNumber()); + dm.setTopicName(cmd.getTopicName()); + dm.setMessage(cmd.getMessageHex()); + dm.setDataType(cmd.getDataType()); + dm.setDeferIfOffline(false); + deviceMessageService.dispatchFunctionGet(dm); + deviceDeferredCommandMapper.updateStatus(cmd.getId(), DeviceDeferredCommand.STATUS_DONE, null); + } catch (Exception e) { + log.warn("=>离线待下发指令执行失败 id={}, serial={}", cmd.getId(), cmd.getSerialNumber(), e); + String msg = e.getMessage(); + if (msg != null && msg.length() > 500) { + msg = msg.substring(0, 500); + } + deviceDeferredCommandMapper.updateStatus(cmd.getId(), DeviceDeferredCommand.STATUS_FAILED, msg); + } + } + } + + private static String normalizeMessagePayload(Object message) { + if (message == null) { + return ""; + } + return message.toString().trim(); + } +} diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java index 6136bbd0..009c6e8b 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceMessageServiceImpl.java @@ -6,6 +6,7 @@ import com.fastbee.common.core.device.DeviceAndProtocol; import com.fastbee.common.core.mq.DeviceReportBo; import com.fastbee.common.core.mq.MQSendMessageBo; import com.fastbee.common.core.mq.message.DeviceMessage; +import com.fastbee.common.core.mq.message.DeviceMessagePostResult; import com.fastbee.common.core.mq.message.FunctionCallBackBo; import com.fastbee.common.core.mq.message.ModbusPollMsg; import com.fastbee.common.core.protocol.modbus.ModbusCode; @@ -23,6 +24,7 @@ import com.fastbee.common.utils.gateway.mq.TopicsUtils; import com.fastbee.common.utils.gateway.protocol.ByteUtils; import com.fastbee.common.utils.modbus.ModbusUtils; import com.fastbee.iot.cache.ITSLCache; +import com.fastbee.iot.data.service.IDeviceDeferredCommandService; import com.fastbee.iot.data.service.IDeviceMessageService; import com.fastbee.iot.domain.ModbusConfig; import com.fastbee.iot.domain.ModbusJob; @@ -45,6 +47,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufUtil; import io.netty.util.ReferenceCountUtil; import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import javax.annotation.Resource; @@ -80,8 +83,12 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService { @Resource private MessageManager messageManager; + @Lazy + @Resource + private IDeviceDeferredCommandService deviceDeferredCommandService; + @Override - public void messagePost(DeviceMessage deviceMessage) { + public DeviceMessagePostResult messagePost(DeviceMessage deviceMessage) { String topicName = deviceMessage.getTopicName(); String serialNumber = deviceMessage.getSerialNumber(); DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(serialNumber); @@ -90,27 +97,14 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService { topicName = topicsUtils.buildTopic(deviceAndProtocol.getProductId(), serialNumber, type); switch (type) { case FUNCTION_GET: - if (transport.equals(ServerType.MQTT.getCode())) { - mqttClient.publish(0, false, topicName, deviceMessage.getMessage().toString()); - } else if (transport.equals(ServerType.TCP.getCode())) { - // 处理TCP下发 - ModbusPollMsg modbusPollMsg = new ModbusPollMsg(); - modbusPollMsg.setSerialNumber(serialNumber); - modbusPollMsg.setProductId(deviceAndProtocol.getProductId()); - List commandList = new ArrayList<>(); - commandList.add(deviceMessage.getMessage().toString()); - modbusPollMsg.setCommandList(commandList); - DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(modbusPollMsg.getSerialNumber()); - modbusPollMsg.setTransport(deviceStatusVO.getTransport()); - if (deviceStatusVO.getStatus() != DeviceStatus.ONLINE.getType()) { - log.info("设备：[{}],不在线", modbusPollMsg.getSerialNumber()); - return; + if (Boolean.TRUE.equals(deviceMessage.getDeferIfOffline())) { + DeviceStatusVO onlineCheck = deviceService.selectDeviceStatusAndTransportStatus(serialNumber); + if (onlineCheck.getStatus() != DeviceStatus.ONLINE.getType()) { + Long taskId = deviceDeferredCommandService.enqueueFunctionGet(deviceMessage); + return DeviceMessagePostResult.deferred(taskId); } - ModbusPollJob modbusPollJob = new ModbusPollJob(); - modbusPollJob.setPollMsg(modbusPollMsg); - modbusPollJob.setType(1); - MessageProducer.sendPropFetch(modbusPollJob); } + dispatchFunctionGetInternal(deviceMessage, deviceAndProtocol, transport, topicName); break; case PROPERTY_POST: //下发的不经过mqtt或TCP直接转发到数据处理模块 @@ -136,6 +130,41 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService { MessageProducer.sendPublishMsg(reportBo); break; } + return DeviceMessagePostResult.sent(); + } + + @Override + public void dispatchFunctionGet(DeviceMessage deviceMessage) { + String serialNumber = deviceMessage.getSerialNumber(); + DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(serialNumber); + String transport = deviceAndProtocol.getTransport(); + TopicType type = TopicType.getType(deviceMessage.getTopicName()); + String topicName = topicsUtils.buildTopic(deviceAndProtocol.getProductId(), serialNumber, type); + dispatchFunctionGetInternal(deviceMessage, deviceAndProtocol, transport, topicName); + } + + private void dispatchFunctionGetInternal(DeviceMessage deviceMessage, DeviceAndProtocol deviceAndProtocol, String transport, String topicName) { + String serialNumber = deviceMessage.getSerialNumber(); + if (transport.equals(ServerType.MQTT.getCode())) { + mqttClient.publish(0, false, topicName, deviceMessage.getMessage().toString()); + } else if (transport.equals(ServerType.TCP.getCode())) { + ModbusPollMsg modbusPollMsg = new ModbusPollMsg(); + modbusPollMsg.setSerialNumber(serialNumber); + modbusPollMsg.setProductId(deviceAndProtocol.getProductId()); + List commandList = new ArrayList<>(); + commandList.add(deviceMessage.getMessage().toString()); + modbusPollMsg.setCommandList(commandList); + DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(modbusPollMsg.getSerialNumber()); + modbusPollMsg.setTransport(deviceStatusVO.getTransport()); + if (deviceStatusVO.getStatus() != DeviceStatus.ONLINE.getType()) { + log.info("设备：[{}],不在线", modbusPollMsg.getSerialNumber()); + return; + } + ModbusPollJob modbusPollJob = new ModbusPollJob(); + modbusPollJob.setPollMsg(modbusPollMsg); + modbusPollJob.setType(1); + MessageProducer.sendPropFetch(modbusPollJob); + } } @Override diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java index e425368f..6ef55528 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java @@ -121,7 +121,7 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi reportMessage.setProductId(productId); reportMessage.setPlatformDate(bo.getPlatformDate()); reportMessage.setServerType(bo.getServerType()); - //同步设备状态 + // 同步设备状态 this.synchDeviceStatus(serialNumber); //处理网关设备上报数据 processNoSub(reportMessage, bo.getTopicName()); @@ -147,7 +147,7 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi //log.debug("TCP数据上报-授权码校验通过: 设备[{}]已绑定授权码", serialNumber); } - //同步设备状态 + // 同步设备状态 this.synchDeviceStatus(serialNumber); DeviceReport deviceReport = new DeviceReport(); BeanUtils.copyProperties(bo, deviceReport); diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/device/DeviceMessageController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/device/DeviceMessageController.java index 21c44af6..047bea45 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/device/DeviceMessageController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/device/DeviceMessageController.java @@ -3,6 +3,7 @@ package com.fastbee.controller.device; import com.fastbee.common.core.controller.BaseController; import com.fastbee.common.core.domain.AjaxResult; import com.fastbee.common.core.mq.message.DeviceMessage; +import com.fastbee.common.core.mq.message.DeviceMessagePostResult; import com.fastbee.iot.data.service.IDeviceMessageService; import com.fastbee.modbus.model.ModbusRtu; import io.swagger.annotations.Api; @@ -24,7 +25,10 @@ public class DeviceMessageController extends BaseController { @PostMapping(value = "/post") @ApiOperation(value = "平台下发指令") public AjaxResult messagePost(@RequestBody DeviceMessage deviceMessage) { - deviceMessageService.messagePost(deviceMessage); + DeviceMessagePostResult result = deviceMessageService.messagePost(deviceMessage); + if (result.isDeferred()) { + return AjaxResult.success("设备离线，已加入上线后自动下发队列").put("deferred", true).put("taskId", result.getTaskId()); + } return AjaxResult.success(); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/DeviceDeferredCommand.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/DeviceDeferredCommand.java new file mode 100644 index 00000000..69676e7b --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/DeviceDeferredCommand.java @@ -0,0 +1,43 @@ +package com.fastbee.iot.domain; + +import com.fasterxml.jackson.annotation.JsonFormat; +import lombok.Data; + +import java.util.Date; + +/** + * 设备离线待下发指令 iot_device_deferred_command + */ +@Data +public class DeviceDeferredCommand { + + public static final int TASK_TYPE_FUNCTION_GET = 1; + + // 待执行 + public static final int STATUS_PENDING = 0; + // 执行中 + public static final int STATUS_DISPATCHING = 1; + // 执行成功 + public static final int STATUS_DONE = 2; + // 执行失败 + public static final int STATUS_FAILED = 3; + + private Long id; + private String serialNumber; + private String topicName; + private String messageHex; + private String dataType; + private Integer taskType; + private Integer status; + private String payloadHash; + private String errMsg; + + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date createTime; + + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date expireTime; + + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date dispatchTime; +} diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceDeferredCommandMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceDeferredCommandMapper.java new file mode 100644 index 00000000..7669abcd --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/mapper/DeviceDeferredCommandMapper.java @@ -0,0 +1,26 @@ +package com.fastbee.iot.mapper; + +import com.fastbee.iot.domain.DeviceDeferredCommand; +import org.apache.ibatis.annotations.Param; +import org.springframework.stereotype.Repository; + +import java.util.List; + +/** + * 设备离线待下发指令 + */ +@Repository +public interface DeviceDeferredCommandMapper { + + int insert(DeviceDeferredCommand row); + + DeviceDeferredCommand selectPendingBySerialAndHash(@Param("serialNumber") String serialNumber, + @Param("payloadHash") String payloadHash); + + List selectPendingListBySerialNumber(@Param("serialNumber") String serialNumber); + + int updateStatus(@Param("id") Long id, @Param("status") Integer status, + @Param("errMsg") String errMsg); + + int updateDispatchingIfPending(@Param("id") Long id); +} diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceDeferredCommandMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceDeferredCommandMapper.xml new file mode 100644 index 00000000..1522373b --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/DeviceDeferredCommandMapper.xml @@ -0,0 +1,61 @@ + + + + + + + + + + + + + + + + + + + + + INSERT INTO iot_device_deferred_command + (serial_number, topic_name, message_hex, data_type, task_type, status, payload_hash, err_msg, create_time, expire_time, dispatch_time) + VALUES + (#{serialNumber}, #{topicName}, #{messageHex}, #{dataType}, #{taskType}, #{status}, #{payloadHash}, #{errMsg}, #{createTime}, #{expireTime}, #{dispatchTime}) + + + + + + + + UPDATE iot_device_deferred_command + SET status = #{status}, + err_msg = #{errMsg} + + , dispatch_time = NOW() + + WHERE id = #{id} + + + + UPDATE iot_device_deferred_command + SET status = 1 + WHERE id = #{id} AND status = 0 + + diff --git a/sql/iot_device_deferred_command.sql b/sql/iot_device_deferred_command.sql new file mode 100644 index 00000000..82b56ba8 --- /dev/null +++ b/sql/iot_device_deferred_command.sql @@ -0,0 +1,20 @@ +-- 离线待下发指令（设备上线后自动执行 FUNCTION_GET 等） +-- 执行：在目标库中运行本脚本一次 + +CREATE TABLE IF NOT EXISTS `iot_device_deferred_command` ( + `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键', + `serial_number` varchar(64) NOT NULL COMMENT '设备编号', + `topic_name` varchar(255) NOT NULL COMMENT '原始 topic，如 /function/get', + `message_hex` varchar(8192) DEFAULT NULL COMMENT '下发内容（十六进制字符串）', + `data_type` varchar(32) DEFAULT NULL COMMENT '数据类型，如 hex', + `task_type` tinyint NOT NULL DEFAULT '1' COMMENT '任务类型：1=FUNCTION_GET', + `status` tinyint NOT NULL DEFAULT '0' COMMENT '0=待执行 1=执行中 2=成功 3=失败', + `payload_hash` varchar(64) NOT NULL COMMENT '去重哈希', + `err_msg` varchar(512) DEFAULT NULL COMMENT '失败原因', + `create_time` datetime DEFAULT NULL COMMENT '创建时间', + `expire_time` datetime DEFAULT NULL COMMENT '过期时间', + `dispatch_time` datetime DEFAULT NULL COMMENT '实际下发时间', + PRIMARY KEY (`id`), + KEY `idx_serial_status` (`serial_number`,`status`), + KEY `idx_serial_hash_pending` (`serial_number`,`payload_hash`,`status`,`expire_time`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='设备离线待下发指令'; diff --git a/sql/yack-20260316.sql b/sql/yack-20260316.sql index 32c8bfdb..cffa352a 100644 --- a/sql/yack-20260316.sql +++ b/sql/yack-20260316.sql @@ -1472,6 +1472,24 @@ CREATE TABLE IF NOT EXISTS `iot_push_user` ( KEY `idx_config_id` (`config_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='推送接收用户'; +-- 离线待下发指令（设备上线后自动执行 FUNCTION_GET 等） +CREATE TABLE IF NOT EXISTS `iot_device_deferred_command` ( + `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键', + `serial_number` varchar(64) NOT NULL COMMENT '设备编号', + `topic_name` varchar(255) NOT NULL COMMENT '原始 topic，如 /function/get', + `message_hex` varchar(8192) DEFAULT NULL COMMENT '下发内容（十六进制字符串）', + `data_type` varchar(32) DEFAULT NULL COMMENT '数据类型，如 hex', + `task_type` tinyint NOT NULL DEFAULT '1' COMMENT '任务类型：1=FUNCTION_GET', + `status` tinyint NOT NULL DEFAULT '0' COMMENT '0=待执行 1=执行中 2=成功 3=失败', + `payload_hash` varchar(64) NOT NULL COMMENT '去重哈希', + `err_msg` varchar(512) DEFAULT NULL COMMENT '失败原因', + `create_time` datetime DEFAULT NULL COMMENT '创建时间', + `expire_time` datetime DEFAULT NULL COMMENT '过期时间', + `dispatch_time` datetime DEFAULT NULL COMMENT '实际下发时间', + PRIMARY KEY (`id`), + KEY `idx_serial_status` (`serial_number`,`status`), + KEY `idx_serial_hash_pending` (`serial_number`,`payload_hash`,`status`,`expire_time`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='设备离线待下发指令'; -- fastbee.iot_var_temp definition diff --git a/yack-web/src/views/iot/device/running-status.vue b/yack-web/src/views/iot/device/running-status.vue index 35904009..37e1ae1d 100644 --- a/yack-web/src/views/iot/device/running-status.vue +++ b/yack-web/src/views/iot/device/running-status.vue @@ -34,7 +34,7 @@ 编辑指令 - 从设备读取 + 从设备读取 @@ -51,7 +51,7 @@ 编辑指令 - 从设备读取 + 从设备读取 @@ -68,7 +68,7 @@ 编辑指令 - + 从设备读取 @@ -724,13 +724,18 @@ export default { const trimmed = String(cmd).trim(); const hexMessage = this.isHexLike(trimmed) ? this.normalizeHexString(trimmed) : this.asciiCommandToHex(trimmed); - await messagePost({ + const res = await messagePost({ message: hexMessage, serialNumber: this.deviceInfo.serialNumber, topicName: '/function/get', dataType: 'hex', + deferIfOffline: true, }); - this.$message({ type: 'success', message: '已下发指令，等待设备回复解析…' }); + if (res && res.deferred) { + this.$message({ type: 'warning', message: res.msg || '设备离线，已加入上线后自动下发队列' }); + } else { + this.$message({ type: 'success', message: '已下发指令，等待设备回复解析…' }); + } // 后端收到回复后会自动解析并写入数据库，这里做一次轻量轮询刷新3秒、5秒各一次 setTimeout(() => this.refreshDeviceBaseInfo(), 3000);