20260412:运行指令支持设备离线触发(离线-记录触发任务)
This commit is contained in:
parent
cdaf411ea1
commit
8c9fe1e355
@ -219,6 +219,12 @@ liteflow:
|
|||||||
scriptEnableField: enable
|
scriptEnableField: enable
|
||||||
|
|
||||||
|
|
||||||
|
# 离线待下发指令(设备上线后自动补发 FUNCTION_GET)
|
||||||
|
iot:
|
||||||
|
deferred-command:
|
||||||
|
expire-hours: 72
|
||||||
|
flush-delay-ms: 1000
|
||||||
|
|
||||||
# 地图配置
|
# 地图配置
|
||||||
map:
|
map:
|
||||||
provider: gaode # 提供者
|
provider: gaode # 提供者
|
||||||
|
|||||||
@ -32,6 +32,11 @@ public class DeviceMessage {
|
|||||||
|
|
||||||
private String dataType;
|
private String dataType;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 为 true 且设备离线时,将 FUNCTION_GET 写入待执行队列,等设备上线后再下发
|
||||||
|
*/
|
||||||
|
private Boolean deferIfOffline;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 时间
|
* 时间
|
||||||
*/
|
*/
|
||||||
|
|||||||
@ -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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -132,7 +132,7 @@ public class DevicePropFetchConsumer {
|
|||||||
modbusJob.setStatus("0");
|
modbusJob.setStatus("0");
|
||||||
List<ModbusJob> modbusJobList = modbusJobService.selectModbusJobList(modbusJob);
|
List<ModbusJob> modbusJobList = modbusJobService.selectModbusJobList(modbusJob);
|
||||||
if (CollectionUtils.isEmpty(modbusJobList)) return modbusPollMsg;
|
if (CollectionUtils.isEmpty(modbusJobList)) return modbusPollMsg;
|
||||||
//处理子设备情况
|
// 处理子设备情况
|
||||||
handleSubDeviceStatus(modbusJobList);
|
handleSubDeviceStatus(modbusJobList);
|
||||||
List<String> commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList());
|
List<String> commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList());
|
||||||
|
|
||||||
|
|||||||
@ -10,6 +10,7 @@ import com.fastbee.common.core.mq.message.ReportDataBo;
|
|||||||
import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
|
import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
|
||||||
import com.fastbee.common.enums.DeviceStatus;
|
import com.fastbee.common.enums.DeviceStatus;
|
||||||
import com.fastbee.iot.cache.IDeviceCache;
|
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.IMqttMessagePublish;
|
||||||
import com.fastbee.iot.data.service.IRuleEngine;
|
import com.fastbee.iot.data.service.IRuleEngine;
|
||||||
import com.fastbee.iot.domain.Device;
|
import com.fastbee.iot.domain.Device;
|
||||||
@ -57,6 +58,8 @@ public class DeviceStatusConsumer {
|
|||||||
private Boolean enabled;
|
private Boolean enabled;
|
||||||
@Resource
|
@Resource
|
||||||
private ISubGatewayService subGatewayService;
|
private ISubGatewayService subGatewayService;
|
||||||
|
@Resource
|
||||||
|
private IDeviceDeferredCommandService deviceDeferredCommandService;
|
||||||
private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(2);
|
private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(2);
|
||||||
|
|
||||||
public synchronized void consume(DeviceStatusBo bo) {
|
public synchronized void consume(DeviceStatusBo bo) {
|
||||||
@ -90,9 +93,14 @@ public class DeviceStatusConsumer {
|
|||||||
}
|
}
|
||||||
/*更新设备状态*/
|
/*更新设备状态*/
|
||||||
device.setProtocolCode(product.getProtocolCode());
|
device.setProtocolCode(product.getProtocolCode());
|
||||||
|
boolean transitionToOnline = status.equals(DeviceStatus.ONLINE)
|
||||||
|
&& device.getStatus() != DeviceStatus.ONLINE.getType();
|
||||||
deviceCache.updateDeviceStatusCache(statusBo, device);
|
deviceCache.updateDeviceStatusCache(statusBo, device);
|
||||||
//处理影子模式值
|
//处理影子模式值
|
||||||
this.handlerShadow(device, status);
|
this.handlerShadow(device, status);
|
||||||
|
if (transitionToOnline) {
|
||||||
|
deviceDeferredCommandService.flushPendingOnOnline(statusBo.getSerialNumber());
|
||||||
|
}
|
||||||
//设备上下线执行规则引擎
|
//设备上下线执行规则引擎
|
||||||
ReportDataBo dataBo = new ReportDataBo();
|
ReportDataBo dataBo = new ReportDataBo();
|
||||||
dataBo.setRuleEngine(true);
|
dataBo.setRuleEngine(true);
|
||||||
|
|||||||
@ -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);
|
||||||
|
}
|
||||||
@ -1,6 +1,7 @@
|
|||||||
package com.fastbee.iot.data.service;
|
package com.fastbee.iot.data.service;
|
||||||
|
|
||||||
import com.fastbee.common.core.mq.message.DeviceMessage;
|
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.common.core.thingsModel.ThingsModelSimpleItem;
|
||||||
import com.fastbee.iot.model.VariableReadVO;
|
import com.fastbee.iot.model.VariableReadVO;
|
||||||
import com.fastbee.modbus.model.ModbusRtu;
|
import com.fastbee.modbus.model.ModbusRtu;
|
||||||
@ -12,7 +13,15 @@ import java.util.List;
|
|||||||
*/
|
*/
|
||||||
public interface IDeviceMessageService {
|
public interface IDeviceMessageService {
|
||||||
|
|
||||||
void messagePost(DeviceMessage deviceMessage);
|
/**
|
||||||
|
* 平台下发指令(支持 deferIfOffline 离线入队)
|
||||||
|
*/
|
||||||
|
DeviceMessagePostResult messagePost(DeviceMessage deviceMessage);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 直接下发 FUNCTION_GET(内部与上线补发使用,不再做离线入队判断)
|
||||||
|
*/
|
||||||
|
void dispatchFunctionGet(DeviceMessage deviceMessage);
|
||||||
|
|
||||||
String messageEncode(ModbusRtu modbusRtu);
|
String messageEncode(ModbusRtu modbusRtu);
|
||||||
|
|
||||||
|
|||||||
@ -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<DeviceDeferredCommand> 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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -6,6 +6,7 @@ import com.fastbee.common.core.device.DeviceAndProtocol;
|
|||||||
import com.fastbee.common.core.mq.DeviceReportBo;
|
import com.fastbee.common.core.mq.DeviceReportBo;
|
||||||
import com.fastbee.common.core.mq.MQSendMessageBo;
|
import com.fastbee.common.core.mq.MQSendMessageBo;
|
||||||
import com.fastbee.common.core.mq.message.DeviceMessage;
|
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.FunctionCallBackBo;
|
||||||
import com.fastbee.common.core.mq.message.ModbusPollMsg;
|
import com.fastbee.common.core.mq.message.ModbusPollMsg;
|
||||||
import com.fastbee.common.core.protocol.modbus.ModbusCode;
|
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.gateway.protocol.ByteUtils;
|
||||||
import com.fastbee.common.utils.modbus.ModbusUtils;
|
import com.fastbee.common.utils.modbus.ModbusUtils;
|
||||||
import com.fastbee.iot.cache.ITSLCache;
|
import com.fastbee.iot.cache.ITSLCache;
|
||||||
|
import com.fastbee.iot.data.service.IDeviceDeferredCommandService;
|
||||||
import com.fastbee.iot.data.service.IDeviceMessageService;
|
import com.fastbee.iot.data.service.IDeviceMessageService;
|
||||||
import com.fastbee.iot.domain.ModbusConfig;
|
import com.fastbee.iot.domain.ModbusConfig;
|
||||||
import com.fastbee.iot.domain.ModbusJob;
|
import com.fastbee.iot.domain.ModbusJob;
|
||||||
@ -45,6 +47,7 @@ import io.netty.buffer.ByteBuf;
|
|||||||
import io.netty.buffer.ByteBufUtil;
|
import io.netty.buffer.ByteBufUtil;
|
||||||
import io.netty.util.ReferenceCountUtil;
|
import io.netty.util.ReferenceCountUtil;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.context.annotation.Lazy;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
import javax.annotation.Resource;
|
import javax.annotation.Resource;
|
||||||
@ -80,8 +83,12 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
|
|||||||
@Resource
|
@Resource
|
||||||
private MessageManager messageManager;
|
private MessageManager messageManager;
|
||||||
|
|
||||||
|
@Lazy
|
||||||
|
@Resource
|
||||||
|
private IDeviceDeferredCommandService deviceDeferredCommandService;
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void messagePost(DeviceMessage deviceMessage) {
|
public DeviceMessagePostResult messagePost(DeviceMessage deviceMessage) {
|
||||||
String topicName = deviceMessage.getTopicName();
|
String topicName = deviceMessage.getTopicName();
|
||||||
String serialNumber = deviceMessage.getSerialNumber();
|
String serialNumber = deviceMessage.getSerialNumber();
|
||||||
DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(serialNumber);
|
DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(serialNumber);
|
||||||
@ -90,27 +97,14 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
|
|||||||
topicName = topicsUtils.buildTopic(deviceAndProtocol.getProductId(), serialNumber, type);
|
topicName = topicsUtils.buildTopic(deviceAndProtocol.getProductId(), serialNumber, type);
|
||||||
switch (type) {
|
switch (type) {
|
||||||
case FUNCTION_GET:
|
case FUNCTION_GET:
|
||||||
if (transport.equals(ServerType.MQTT.getCode())) {
|
if (Boolean.TRUE.equals(deviceMessage.getDeferIfOffline())) {
|
||||||
mqttClient.publish(0, false, topicName, deviceMessage.getMessage().toString());
|
DeviceStatusVO onlineCheck = deviceService.selectDeviceStatusAndTransportStatus(serialNumber);
|
||||||
} else if (transport.equals(ServerType.TCP.getCode())) {
|
if (onlineCheck.getStatus() != DeviceStatus.ONLINE.getType()) {
|
||||||
// 处理TCP下发
|
Long taskId = deviceDeferredCommandService.enqueueFunctionGet(deviceMessage);
|
||||||
ModbusPollMsg modbusPollMsg = new ModbusPollMsg();
|
return DeviceMessagePostResult.deferred(taskId);
|
||||||
modbusPollMsg.setSerialNumber(serialNumber);
|
|
||||||
modbusPollMsg.setProductId(deviceAndProtocol.getProductId());
|
|
||||||
List<String> 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);
|
|
||||||
}
|
}
|
||||||
|
dispatchFunctionGetInternal(deviceMessage, deviceAndProtocol, transport, topicName);
|
||||||
break;
|
break;
|
||||||
case PROPERTY_POST:
|
case PROPERTY_POST:
|
||||||
//下发的不经过mqtt或TCP直接转发到数据处理模块
|
//下发的不经过mqtt或TCP直接转发到数据处理模块
|
||||||
@ -136,6 +130,41 @@ public class DeviceMessageServiceImpl implements IDeviceMessageService {
|
|||||||
MessageProducer.sendPublishMsg(reportBo);
|
MessageProducer.sendPublishMsg(reportBo);
|
||||||
break;
|
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<String> 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
|
@Override
|
||||||
|
|||||||
@ -121,7 +121,7 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi
|
|||||||
reportMessage.setProductId(productId);
|
reportMessage.setProductId(productId);
|
||||||
reportMessage.setPlatformDate(bo.getPlatformDate());
|
reportMessage.setPlatformDate(bo.getPlatformDate());
|
||||||
reportMessage.setServerType(bo.getServerType());
|
reportMessage.setServerType(bo.getServerType());
|
||||||
//同步设备状态
|
// 同步设备状态
|
||||||
this.synchDeviceStatus(serialNumber);
|
this.synchDeviceStatus(serialNumber);
|
||||||
//处理网关设备上报数据
|
//处理网关设备上报数据
|
||||||
processNoSub(reportMessage, bo.getTopicName());
|
processNoSub(reportMessage, bo.getTopicName());
|
||||||
@ -147,7 +147,7 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi
|
|||||||
//log.debug("TCP数据上报-授权码校验通过: 设备[{}]已绑定授权码", serialNumber);
|
//log.debug("TCP数据上报-授权码校验通过: 设备[{}]已绑定授权码", serialNumber);
|
||||||
}
|
}
|
||||||
|
|
||||||
//同步设备状态
|
// 同步设备状态
|
||||||
this.synchDeviceStatus(serialNumber);
|
this.synchDeviceStatus(serialNumber);
|
||||||
DeviceReport deviceReport = new DeviceReport();
|
DeviceReport deviceReport = new DeviceReport();
|
||||||
BeanUtils.copyProperties(bo, deviceReport);
|
BeanUtils.copyProperties(bo, deviceReport);
|
||||||
|
|||||||
@ -3,6 +3,7 @@ package com.fastbee.controller.device;
|
|||||||
import com.fastbee.common.core.controller.BaseController;
|
import com.fastbee.common.core.controller.BaseController;
|
||||||
import com.fastbee.common.core.domain.AjaxResult;
|
import com.fastbee.common.core.domain.AjaxResult;
|
||||||
import com.fastbee.common.core.mq.message.DeviceMessage;
|
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.iot.data.service.IDeviceMessageService;
|
||||||
import com.fastbee.modbus.model.ModbusRtu;
|
import com.fastbee.modbus.model.ModbusRtu;
|
||||||
import io.swagger.annotations.Api;
|
import io.swagger.annotations.Api;
|
||||||
@ -24,7 +25,10 @@ public class DeviceMessageController extends BaseController {
|
|||||||
@PostMapping(value = "/post")
|
@PostMapping(value = "/post")
|
||||||
@ApiOperation(value = "平台下发指令")
|
@ApiOperation(value = "平台下发指令")
|
||||||
public AjaxResult messagePost(@RequestBody DeviceMessage deviceMessage) {
|
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();
|
return AjaxResult.success();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -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;
|
||||||
|
}
|
||||||
@ -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<DeviceDeferredCommand> 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);
|
||||||
|
}
|
||||||
@ -0,0 +1,61 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8" ?>
|
||||||
|
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
|
||||||
|
<mapper namespace="com.fastbee.iot.mapper.DeviceDeferredCommandMapper">
|
||||||
|
|
||||||
|
<resultMap type="com.fastbee.iot.domain.DeviceDeferredCommand" id="DeviceDeferredCommandResult">
|
||||||
|
<result property="id" column="id"/>
|
||||||
|
<result property="serialNumber" column="serial_number"/>
|
||||||
|
<result property="topicName" column="topic_name"/>
|
||||||
|
<result property="messageHex" column="message_hex"/>
|
||||||
|
<result property="dataType" column="data_type"/>
|
||||||
|
<result property="taskType" column="task_type"/>
|
||||||
|
<result property="status" column="status"/>
|
||||||
|
<result property="payloadHash" column="payload_hash"/>
|
||||||
|
<result property="errMsg" column="err_msg"/>
|
||||||
|
<result property="createTime" column="create_time"/>
|
||||||
|
<result property="expireTime" column="expire_time"/>
|
||||||
|
<result property="dispatchTime" column="dispatch_time"/>
|
||||||
|
</resultMap>
|
||||||
|
|
||||||
|
<insert id="insert" parameterType="com.fastbee.iot.domain.DeviceDeferredCommand" useGeneratedKeys="true" keyProperty="id">
|
||||||
|
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})
|
||||||
|
</insert>
|
||||||
|
|
||||||
|
<select id="selectPendingBySerialAndHash" resultMap="DeviceDeferredCommandResult">
|
||||||
|
SELECT id, serial_number, topic_name, message_hex, data_type, task_type, status, payload_hash, err_msg, create_time, expire_time, dispatch_time
|
||||||
|
FROM iot_device_deferred_command
|
||||||
|
WHERE serial_number = #{serialNumber}
|
||||||
|
AND payload_hash = #{payloadHash}
|
||||||
|
AND status = 0
|
||||||
|
AND (expire_time IS NULL OR expire_time > NOW())
|
||||||
|
LIMIT 1
|
||||||
|
</select>
|
||||||
|
|
||||||
|
<select id="selectPendingListBySerialNumber" resultMap="DeviceDeferredCommandResult">
|
||||||
|
SELECT id, serial_number, topic_name, message_hex, data_type, task_type, status, payload_hash, err_msg, create_time, expire_time, dispatch_time
|
||||||
|
FROM iot_device_deferred_command
|
||||||
|
WHERE serial_number = #{serialNumber}
|
||||||
|
AND status = 0
|
||||||
|
AND (expire_time IS NULL OR expire_time > NOW())
|
||||||
|
ORDER BY create_time ASC, id ASC
|
||||||
|
</select>
|
||||||
|
|
||||||
|
<update id="updateStatus">
|
||||||
|
UPDATE iot_device_deferred_command
|
||||||
|
SET status = #{status},
|
||||||
|
err_msg = #{errMsg}
|
||||||
|
<if test="status != null and (status == 2 or status == 3)">
|
||||||
|
, dispatch_time = NOW()
|
||||||
|
</if>
|
||||||
|
WHERE id = #{id}
|
||||||
|
</update>
|
||||||
|
|
||||||
|
<update id="updateDispatchingIfPending">
|
||||||
|
UPDATE iot_device_deferred_command
|
||||||
|
SET status = 1
|
||||||
|
WHERE id = #{id} AND status = 0
|
||||||
|
</update>
|
||||||
|
</mapper>
|
||||||
20
sql/iot_device_deferred_command.sql
Normal file
20
sql/iot_device_deferred_command.sql
Normal file
@ -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='设备离线待下发指令';
|
||||||
@ -1472,6 +1472,24 @@ CREATE TABLE IF NOT EXISTS `iot_push_user` (
|
|||||||
KEY `idx_config_id` (`config_id`)
|
KEY `idx_config_id` (`config_id`)
|
||||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='推送接收用户';
|
) 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
|
-- fastbee.iot_var_temp definition
|
||||||
|
|||||||
@ -34,7 +34,7 @@
|
|||||||
<el-input v-model="imeiCmd" size="mini" placeholder="例如:AT+IMEI\\r\\n (回车下发)" @keyup.enter.native="handleEnterSend('imei')" />
|
<el-input v-model="imeiCmd" size="mini" placeholder="例如:AT+IMEI\\r\\n (回车下发)" @keyup.enter.native="handleEnterSend('imei')" />
|
||||||
<el-button slot="reference" size="mini" type="default">编辑指令</el-button>
|
<el-button slot="reference" size="mini" type="default">编辑指令</el-button>
|
||||||
</el-popover>
|
</el-popover>
|
||||||
<el-button type="primary" size="mini" :disabled="deviceInfo.status != 3" @click="sendAt('imei')" :loading="imeiLoading" title="默认指令:AT+IMEI\r\n">从设备读取</el-button>
|
<el-button type="primary" size="mini" @click="sendAt('imei')" :loading="imeiLoading" title="默认指令:AT+IMEI\r\n">从设备读取</el-button>
|
||||||
</div>
|
</div>
|
||||||
</el-descriptions-item>
|
</el-descriptions-item>
|
||||||
|
|
||||||
@ -51,7 +51,7 @@
|
|||||||
<el-input v-model="iccidCmd" size="mini" placeholder="例如:AT+ICCID\\r\\n (回车下发)" @keyup.enter.native="handleEnterSend('iccid')" />
|
<el-input v-model="iccidCmd" size="mini" placeholder="例如:AT+ICCID\\r\\n (回车下发)" @keyup.enter.native="handleEnterSend('iccid')" />
|
||||||
<el-button slot="reference" size="mini" type="default">编辑指令</el-button>
|
<el-button slot="reference" size="mini" type="default">编辑指令</el-button>
|
||||||
</el-popover>
|
</el-popover>
|
||||||
<el-button type="primary" size="mini" :disabled="deviceInfo.status != 3" @click="sendAt('iccid')" :loading="iccidLoading" title="默认指令:AT+ICCID\r\n">从设备读取</el-button>
|
<el-button type="primary" size="mini" @click="sendAt('iccid')" :loading="iccidLoading" title="默认指令:AT+ICCID\r\n">从设备读取</el-button>
|
||||||
</div>
|
</div>
|
||||||
</el-descriptions-item>
|
</el-descriptions-item>
|
||||||
|
|
||||||
@ -68,7 +68,7 @@
|
|||||||
<el-input v-model="lbsCmd" size="mini" placeholder="例如:AT+LBS\\r\\n 或 Hex:41 54 2b 4c 42 53 0d 0a (回车下发)" @keyup.enter.native="handleEnterSend('lbs')" />
|
<el-input v-model="lbsCmd" size="mini" placeholder="例如:AT+LBS\\r\\n 或 Hex:41 54 2b 4c 42 53 0d 0a (回车下发)" @keyup.enter.native="handleEnterSend('lbs')" />
|
||||||
<el-button slot="reference" size="mini" type="default">编辑指令</el-button>
|
<el-button slot="reference" size="mini" type="default">编辑指令</el-button>
|
||||||
</el-popover>
|
</el-popover>
|
||||||
<el-button type="primary" size="mini" :disabled="deviceInfo.status != 3" @click="sendAt('lbs')" :loading="lbsLoading" title="默认指令:AT+LBS\r\n(ASCII 转 Hex 为 41542B4C42530D0A)">
|
<el-button type="primary" size="mini" @click="sendAt('lbs')" :loading="lbsLoading" title="默认指令:AT+LBS\r\n(ASCII 转 Hex 为 41542B4C42530D0A)">
|
||||||
从设备读取
|
从设备读取
|
||||||
</el-button>
|
</el-button>
|
||||||
</div>
|
</div>
|
||||||
@ -724,13 +724,18 @@ export default {
|
|||||||
const trimmed = String(cmd).trim();
|
const trimmed = String(cmd).trim();
|
||||||
const hexMessage = this.isHexLike(trimmed) ? this.normalizeHexString(trimmed) : this.asciiCommandToHex(trimmed);
|
const hexMessage = this.isHexLike(trimmed) ? this.normalizeHexString(trimmed) : this.asciiCommandToHex(trimmed);
|
||||||
|
|
||||||
await messagePost({
|
const res = await messagePost({
|
||||||
message: hexMessage,
|
message: hexMessage,
|
||||||
serialNumber: this.deviceInfo.serialNumber,
|
serialNumber: this.deviceInfo.serialNumber,
|
||||||
topicName: '/function/get',
|
topicName: '/function/get',
|
||||||
dataType: 'hex',
|
dataType: 'hex',
|
||||||
|
deferIfOffline: true,
|
||||||
});
|
});
|
||||||
|
if (res && res.deferred) {
|
||||||
|
this.$message({ type: 'warning', message: res.msg || '设备离线,已加入上线后自动下发队列' });
|
||||||
|
} else {
|
||||||
this.$message({ type: 'success', message: '已下发指令,等待设备回复解析…' });
|
this.$message({ type: 'success', message: '已下发指令,等待设备回复解析…' });
|
||||||
|
}
|
||||||
|
|
||||||
// 后端收到回复后会自动解析并写入数据库,这里做一次轻量轮询刷新3秒、5秒各一次
|
// 后端收到回复后会自动解析并写入数据库,这里做一次轻量轮询刷新3秒、5秒各一次
|
||||||
setTimeout(() => this.refreshDeviceBaseInfo(), 3000);
|
setTimeout(() => this.refreshDeviceBaseInfo(), 3000);
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user