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

fix(app): 修复 AppController 安全与查询问题

Browse Source 
- 加强异常处理、类型安全和图片上传校验
- 优化设备相关查询，避免重复访问数据源
- 补充并记录并发测试与审查修复实施计划
This commit is contained in:
yuhaiming
2026-07-17 08:20:44 +08:00
parent 4a4aabf19e
commit 70e2356f22
82 changed files with 6070 additions and 583 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,11 +39,6 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
private final AppDeviceMapper appDeviceMapper;
private static final String DEVICE_STATUS_LOCK_PREFIX = "lock:mqtt:device:status:";
public void savePending(DeviceCommand command) {
RedisUtils.setCacheObject(pendingKey(command.getCommandId()), command, Duration.ofSeconds(mqttProperties.getCommandAck().getPendingTtlSeconds()));
pendingIds().add(command.getCommandId());
}
static boolean ackMatchesPendingCommand(String topicDeviceNo, DeviceCommand pendingCommand) {
return StringUtils.isNotBlank(topicDeviceNo)
&& pendingCommand != null
@@ -68,16 +63,6 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
return true;
}
public void removePending(String commandId) {
if (StringUtils.isBlank(commandId)) {
return;
}
withCommandLock(commandId, () -> {
deletePending(commandId);
return true;
});
}
@Override
public void handleAck(String deviceNo, String payload) {
if (StringUtils.isBlank(payload)) {
@@ -111,7 +96,8 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
return;
}
}
boolean ackSaved = withCommandLock(ack.getCommandId(), () -> {
long lockWaitMs = mqttProperties.getCommandAck().getAckLockWaitMs();
CommandLockResult ackResult = withCommandLock(ack.getCommandId(), lockWaitMs, () -> {
DeviceCommand pending = RedisUtils.getCacheObject(pendingKey(ack.getCommandId()));
if (pending == null) {
log.warn("[MQTT] ACK 对应命令不存在或已过期 设备编号={} 命令编号={}", deviceNo, ack.getCommandId());
@@ -126,14 +112,36 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
return true;
});
if (!ackSaved) {
log.warn("[MQTT] ACK 处理未获得命令锁，本次跳过 设备编号={} 命令编号={}", deviceNo, ack.getCommandId());
if (ackResult == CommandLockResult.LOCK_UNAVAILABLE) {
log.warn("[MQTT] ACK 等待命令锁超时，本次跳过 设备编号={} 命令编号={} 等待毫秒={}",
deviceNo, ack.getCommandId(), lockWaitMs);
return;
}
if (ackResult == CommandLockResult.INTERRUPTED) {
return;
}
if (ackResult == CommandLockResult.ACTION_FAILED) {
return;
}
refreshDeviceOnline(deviceNo);
log.info("[MQTT] 收到命令确认 设备编号={} 命令编号={} message={}", deviceNo, ack.getCommandId(), ack.getMessage());
}
public void savePending(DeviceCommand command) {
RedisUtils.setCacheObject(pendingKey(command.getCommandId()), command, Duration.ofSeconds(mqttProperties.getCommandAck().getPendingTtlSeconds()));
pendingIds().add(command.getCommandId());
}
public void removePending(String commandId) {
if (StringUtils.isBlank(commandId)) {
return;
}
withCommandLock(commandId, () -> {
deletePending(commandId);
return true;
});
}
private void handlePlainAck(String deviceNo, String payload) {
List<DeviceCommand> pendingCommands = findPendingCommandsByDeviceNo(deviceNo);
refreshDeviceOnline(deviceNo);
@@ -150,7 +158,8 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
ack.setStatus("1");
ack.setMessage(payload);
boolean ackSaved = withCommandLock(commandId, () -> {
long lockWaitMs = mqttProperties.getCommandAck().getAckLockWaitMs();
CommandLockResult ackResult = withCommandLock(commandId, lockWaitMs, () -> {
DeviceCommand pending = RedisUtils.getCacheObject(pendingKey(commandId));
if (pending == null) {
log.warn("[MQTT] 非 JSON ACK 对应命令不存在或已过期 设备编号={} 命令编号={}", deviceNo, commandId);
@@ -165,13 +174,24 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
return true;
});
if (!ackSaved) {
log.warn("[MQTT] 非 JSON ACK 处理未获得命令锁，本次跳过 设备编号={} 命令编号={}", deviceNo, commandId);
if (ackResult == CommandLockResult.LOCK_UNAVAILABLE) {
log.warn("[MQTT] 非 JSON ACK 等待命令锁超时，本次跳过 设备编号={} 命令编号={} 等待毫秒={}",
deviceNo, commandId, lockWaitMs);
return;
}
if (ackResult == CommandLockResult.INTERRUPTED) {
return;
}
if (ackResult == CommandLockResult.ACTION_FAILED) {
return;
}
log.info("[MQTT] 收到非 JSON 命令确认 设备编号={} 命令编号={} 消息体={}", deviceNo, commandId, payload);
}
private CommandLockResult withCommandLock(String commandId, BooleanSupplier action) {
return withCommandLock(commandId, 0L, action);
}
private List<DeviceCommand> findPendingCommandsByDeviceNo(String deviceNo) {
List<DeviceCommand> commands = new ArrayList<>();
for (String commandId : pendingIds().readAll()) {
@@ -284,22 +304,22 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
pendingIds().remove(commandId);
}
private boolean withCommandLock(String commandId, BooleanSupplier action) {
private CommandLockResult withCommandLock(String commandId, long waitTimeMs, BooleanSupplier action) {
RLock lock = RedisUtils.getClient().getLock(mqttProperties.getCommandAck().getRetryLockKeyPrefix() + commandId);
boolean locked = false;
try {
locked = lock.tryLock(0, mqttProperties.getCommandAck().getRetryLockTtlMs(), TimeUnit.MILLISECONDS);
locked = lock.tryLock(waitTimeMs, mqttProperties.getCommandAck().getRetryLockTtlMs(), TimeUnit.MILLISECONDS);
if (!locked) {
return false;
return CommandLockResult.LOCK_UNAVAILABLE;
}
return action.getAsBoolean();
return action.getAsBoolean() ? CommandLockResult.SUCCESS : CommandLockResult.ACTION_FAILED;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("[MQTT] 命令锁等待被中断 命令编号={}", commandId);
return false;
return CommandLockResult.INTERRUPTED;
} catch (RuntimeException e) {
log.error("[MQTT] 命令锁内处理失败 命令编号={}", commandId, e);
return false;
return CommandLockResult.ACTION_FAILED;
} finally {
if (locked && lock.isHeldByCurrentThread()) {
lock.unlock();
@@ -307,6 +327,13 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
}
}
private enum CommandLockResult {
SUCCESS,
LOCK_UNAVAILABLE,
INTERRUPTED,
ACTION_FAILED
}
private String pendingKey(String commandId) {
return mqttProperties.getCommandAck().getPendingKeyPrefix() + commandId;
}

1
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/config/properties/MqttProperties.java
View File

@@ -64,6 +64,7 @@ public class MqttProperties {
private String pendingSetKey = "mqtt:command:pending:ids";
private String retryLockKeyPrefix = "lock:mqtt:command:retry:";
private long retryLockTtlMs = 30000;
private long ackLockWaitMs = 3000;
private String deviceStatusCachePrefix = "mqtt:device:status:";
private long deviceStatusCacheTtlSeconds = 900;
}

42
water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java
View File

@@ -2,6 +2,7 @@ package org.dromara.mqtt;
import cn.hutool.extra.spring.SpringUtil;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.ibatis.session.Configuration;
import org.dromara.app.domain.AppDevice;
@@ -16,6 +17,7 @@ import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.redisson.api.RLock;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.test.util.ReflectionTestUtils;
@@ -24,6 +26,7 @@ import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
@@ -42,6 +45,7 @@ class MqttCommandAckServiceTest {
}
applicationContext = new GenericApplicationContext();
applicationContext.registerBean(RedissonClient.class, () -> mock(RedissonClient.class));
applicationContext.registerBean(ObjectMapper.class, (Supplier<ObjectMapper>) ObjectMapper::new);
applicationContext.refresh();
new SpringUtil().setApplicationContext(applicationContext);
}
@@ -93,6 +97,44 @@ class MqttCommandAckServiceTest {
assertThat(MqttCommandAckService.ackMatchesPendingCommand("D01", null)).isFalse();
}
@Test
void handleAckWaitsBrieflyForCommandLock() throws Exception {
MqttProperties properties = new MqttProperties();
AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
RedissonClient redissonClient = mock(RedissonClient.class);
RLock commandLock = mock(RLock.class);
RLock statusLock = mock(RLock.class);
RSet<String> pendingIds = mock(RSet.class);
DeviceCommand pending = new DeviceCommand();
pending.setCommandId("cmd-1");
pending.setDeviceNo("D01");
when(redissonClient.getLock("lock:mqtt:command:retry:cmd-1")).thenReturn(commandLock);
when(commandLock.tryLock(3000, 30000, TimeUnit.MILLISECONDS)).thenReturn(true);
when(commandLock.isHeldByCurrentThread()).thenReturn(true);
when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(statusLock);
when(statusLock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(false);
MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(RedisUtils::getClient).thenReturn(redissonClient);
redis.when(() -> RedisUtils.getCacheObject("mqtt:command:pending:cmd-1")).thenReturn(pending);
service.handleAck("D01", "{\"commandId\":\"cmd-1\",\"status\":\"1\"}");
verify(commandLock).tryLock(3000, 30000, TimeUnit.MILLISECONDS);
redis.verify(() -> RedisUtils.setCacheObject(
eq("mqtt:command:ack:cmd-1"),
any(DeviceCommandAck.class),
eq(Duration.ofSeconds(86400))
));
verify(pendingIds).remove("cmd-1");
verify(commandLock).unlock();
}
}
@Test
void refreshDeviceOnlineRenewsStatusCacheTtl() throws Exception {
MqttProperties properties = new MqttProperties();