yuhaiming/water
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

浇水bug修改

Browse Source
This commit is contained in:
yuhaiming
2026-07-09 08:26:37 +08:00
parent 319ebd19ad
commit 7deda9bdc0
27 changed files with 826 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java
View File

@@ -50,6 +50,24 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
withCommandLock(commandId, () -> deletePending(commandId));
}
static boolean resolveMissingCommandId(DeviceCommandAck ack, List<DeviceCommand> pendingCommands) {
if (ack == null || StringUtils.isNotBlank(ack.getCommandId())) {
return ack != null;
}
if (pendingCommands == null || pendingCommands.size() != 1) {
return false;
}
DeviceCommand command = pendingCommands.get(0);
if (command == null || StringUtils.isBlank(command.getCommandId())) {
return false;
}
ack.setCommandId(command.getCommandId());
if (StringUtils.isBlank(ack.getStatus())) {
ack.setStatus("1");
}
return true;
}
@Override
public void handleAck(String deviceNo, String payload) {
if (StringUtils.isBlank(payload)) {
@@ -70,11 +88,19 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
log.warn("[MQTT] ACK JSON 格式错误 设备编号={} 消息体={}", deviceNo, payload, e);
return;
}
if (ack == null || ack.getCommandId() == null) {
if (ack == null) {
log.warn("[MQTT] ACK 缺少命令编号 设备编号={} 消息体={}", deviceNo, payload);
return;
}
ack.setDeviceNo(deviceNo);
if (StringUtils.isBlank(ack.getCommandId())) {
List<DeviceCommand> pendingCommands = findPendingCommandsByDeviceNo(deviceNo);
if (!resolveMissingCommandId(ack, pendingCommands)) {
log.warn("[MQTT] ACK 缺少命令编号且无法唯一匹配待确认命令 设备编号={} 待确认数量={} 消息体={}",
deviceNo, pendingCommands.size(), payload);
return;
}
}
boolean ackSaved = withCommandLock(ack.getCommandId(), () -> {
deletePending(ack.getCommandId());
RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
@@ -84,7 +110,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
return;
}
refreshDeviceOnline(deviceNo);
log.info("[MQTT] 收到命令确认 设备编号={} 命令编号={} 状态={}", deviceNo, ack.getCommandId(), ack.getStatus());
log.info("[MQTT] 收到命令确认 设备编号={} 命令编号={} message={}", deviceNo, ack.getCommandId(), ack.getMessage());
}
private void handlePlainAck(String deviceNo, String payload) {