diff --git a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java index cbe46cc..c3cb3c2 100644 --- a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java +++ b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java @@ -132,6 +132,15 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler { pendingIds().add(command.getCommandId()); } + public int clearPendingCommands() { + Set 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; diff --git a/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java index efd19f0..561bd51 100644 --- a/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java +++ b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java @@ -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 pendingIds = mock(RSet.class); + when(redissonClient.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 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 pendingIds = mock(RSet.class); + when(redissonClient.getSet("mqtt:command:pending:ids")).thenReturn(pendingIds); + when(pendingIds.readAll()).thenReturn(Set.of()); + + MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper); + try (MockedStatic 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();