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

fix: clear mqtt pending commands on request

Browse Source
This commit is contained in:
yuhaiming
2026-07-17 16:20:47 +08:00
parent 1c122dfc68
commit 8b0612f744
2 changed files with 52 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -132,6 +132,15 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
pendingIds().add(command.getCommandId());
}
public int clearPendingCommands() {
Set<String> commandIds = pendingIds().readAll();
for (String commandId : commandIds) {
RedisUtils.deleteObject(pendingKey(commandId));
}
pendingIds().clear();
return commandIds.size();
}
public void removePending(String commandId) {
if (StringUtils.isBlank(commandId)) {
return;

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

@@ -25,6 +25,7 @@ import org.springframework.test.util.ReflectionTestUtils;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
@@ -97,6 +98,48 @@ class MqttCommandAckServiceTest {
assertThat(MqttCommandAckService.ackMatchesPendingCommand("D01", null)).isFalse();
}
@Test
void clearPendingCommandsDeletesPendingObjectsAndClearsIds() {
MqttProperties properties = new MqttProperties();
AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
RedissonClient redissonClient = mock(RedissonClient.class);
RSet<String> pendingIds = mock(RSet.class);
when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
when(pendingIds.readAll()).thenReturn(Set.of("cmd-1", "cmd-2"));
MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(RedisUtils::getClient).thenReturn(redissonClient);
int cleared = service.clearPendingCommands();
assertThat(cleared).isEqualTo(2);
redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1"));
redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-2"));
verify(pendingIds).clear();
}
}
@Test
void clearPendingCommandsReturnsZeroForEmptySet() {
MqttProperties properties = new MqttProperties();
AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
RedissonClient redissonClient = mock(RedissonClient.class);
RSet<String> pendingIds = mock(RSet.class);
when(redissonClient.<String>getSet("mqtt:command:pending:ids")).thenReturn(pendingIds);
when(pendingIds.readAll()).thenReturn(Set.of());
MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(RedisUtils::getClient).thenReturn(redissonClient);
int cleared = service.clearPendingCommands();
assertThat(cleared).isZero();
verify(pendingIds).clear();
}
}
@Test
void handleAckWaitsBrieflyForCommandLock() throws Exception {
MqttProperties properties = new MqttProperties();