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-20 15:45:13 +08:00
parent b6c1edaeba
commit 215d6ba167
20 changed files with 664 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -261,6 +261,14 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
log.warn("[MQTT] 命令重试次数已达上限 设备编号={} 命令编号={}", command.getDeviceNo(), commandId);
return;
}
if (Boolean.FALSE.equals(command.getRetryEnabled())) {
command.setRetryCount(command.getRetryCount() + 1);
command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
savePending(command);
log.debug("[MQTT] 命令已关闭自动重发，等待 ACK 或到期清理 设备编号={} 命令编号={} 等待次数={}",
command.getDeviceNo(), commandId, command.getRetryCount());
return;
}
if (!isDeviceOnline(command.getDeviceNo())) {
command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
savePending(command);

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

@@ -27,8 +27,8 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
@@ -251,6 +251,127 @@ class MqttCommandAckServiceTest {
}
}
@Test
void retryExpiredCommandsAdvancesRetryDisabledCommandsWithoutPublishing() throws Exception {
MqttProperties properties = new MqttProperties();
AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
RedissonClient redissonClient = mock(RedissonClient.class);
RLock commandLock = mock(RLock.class);
RSet<String> pendingIds = mock(RSet.class);
MqttClientManager mqttClientManager = mock(MqttClientManager.class);
DeviceCommand pending = new DeviceCommand();
pending.setCommandId("cmd-1");
pending.setDeviceNo("D01");
pending.setTopic("/aa:bb:cc/subscriber/cmd");
pending.setRetryEnabled(false);
pending.setNextRetryAt(0);
when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
when(pendingIds.readAll()).thenReturn(Set.of("cmd-1"));
when(redissonClient.getLock("lock:mqtt:command:retry:cmd-1")).thenReturn(commandLock);
when(commandLock.tryLock(0, 30000, TimeUnit.MILLISECONDS)).thenReturn(true);
when(commandLock.isHeldByCurrentThread()).thenReturn(true);
MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
ReflectionTestUtils.setField(service, "mqttClientManager", mqttClientManager);
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.retryExpiredCommands();
verify(mqttClientManager, never()).publish(any(String.class), any(String.class));
assertThat(pending.getRetryCount()).isEqualTo(1);
assertThat(pending.getNextRetryAt()).isGreaterThan(0);
redis.verify(() -> RedisUtils.setCacheObject(
eq("mqtt:command:pending:cmd-1"),
eq(pending),
eq(Duration.ofSeconds(86400))
));
verify(pendingIds).add("cmd-1");
redis.verify(() -> RedisUtils.deleteObject(any(String.class)), never());
verify(commandLock).unlock();
}
}
@Test
void retryExpiredCommandsDeletesRetryDisabledCommandsAtRetryLimit() throws Exception {
MqttProperties properties = new MqttProperties();
AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
RedissonClient redissonClient = mock(RedissonClient.class);
RLock commandLock = mock(RLock.class);
RSet<String> pendingIds = mock(RSet.class);
MqttClientManager mqttClientManager = mock(MqttClientManager.class);
DeviceCommand pending = new DeviceCommand();
pending.setCommandId("cmd-1");
pending.setDeviceNo("D01");
pending.setTopic("/aa:bb:cc/subscriber/cmd");
pending.setRetryEnabled(false);
pending.setRetryCount(properties.getCommandAck().getMaxRetryCount());
pending.setNextRetryAt(0);
when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
when(pendingIds.readAll()).thenReturn(Set.of("cmd-1"));
when(redissonClient.getLock("lock:mqtt:command:retry:cmd-1")).thenReturn(commandLock);
when(commandLock.tryLock(0, 30000, TimeUnit.MILLISECONDS)).thenReturn(true);
when(commandLock.isHeldByCurrentThread()).thenReturn(true);
MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
ReflectionTestUtils.setField(service, "mqttClientManager", mqttClientManager);
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.retryExpiredCommands();
verify(mqttClientManager, never()).publish(any(String.class), any(String.class));
redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1"));
verify(pendingIds).remove("cmd-1");
verify(commandLock).unlock();
}
}
@Test
void handleAckDeletesRetryDisabledPendingCommand() 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");
pending.setRetryEnabled(false);
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\"}");
redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1"));
verify(pendingIds).remove("cmd-1");
redis.verify(() -> RedisUtils.setCacheObject(
eq("mqtt:command:ack:cmd-1"),
any(DeviceCommandAck.class),
eq(Duration.ofSeconds(86400))
));
verify(commandLock).unlock();
}
}
@Test
void refreshDeviceOnlineRenewsStatusCacheTtl() throws Exception {
MqttProperties properties = new MqttProperties();