60 lines
6.0 KiB
Markdown
60 lines
6.0 KiB
Markdown
# MQTT Pending Cleanup And Retry Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
|
||
|
||
**Goal:** 单实例应用启动完成后清理旧 MQTT pending 命令，并让新命令等待 10 秒后才进行 ACK 重试。
|
||
|
||
**Architecture:** `MqttCommandAckService` 提供幂等清理方法，独立的 `ApplicationReadyEvent` 监听器在 Redis 初始化完成后调用。命令发送继续由 `DeviceMqttCommandPublisher` 设置 `nextRetryAt`，只调整默认值和配置为 10000 毫秒。
|
||
|
||
**Tech Stack:** Spring Boot、ApplicationReadyEvent、Spring `@EventListener`、Redisson `RSet`、RedisUtils、JUnit 5、Mockito、Maven。
|
||
|
||
## Global Constraints
|
||
- 当前按单实例部署，启动时清理全部 pending 命令。
|
||
- 只删除 pending ID 集合和对应命令缓存，不删除 ACK 历史缓存。
|
||
- 清理失败记录完整异常，不阻止应用启动。
|
||
- `retry-interval-ms` 使用 `10000`，`scan-interval-ms` 保持 `5000`。
|
||
- 最大重试次数保持 3 次，不包含首次发送。
|
||
- 不修改 MQTT Topic、ACK 消息格式和 Payload。
|
||
|
||
---
|
||
|
||
### Task 1: Add Pending Cleanup Behavior
|
||
|
||
**Files:** Modify `water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java`; test `water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java`.
|
||
|
||
**Interface:** add `public int clearPendingCommands()` using existing `pendingIds()`, `pendingKey(String)`, and Redis key conventions.
|
||
|
||
- [ ] Write a failing test: stub `RSet.readAll()` with `cmd-1` and `cmd-2`; call the method; assert return `2`, verify deletion of both `mqtt:command:pending:{id}` objects, and verify `pendingIds.clear()`. Add an empty-set case returning `0`.
|
||
- [ ] Run `mvn -pl water-common/water-common-mqtt -am "-DskipTests=false" "-Dmaven.test.skip=false" "-Dtest=MqttCommandAckServiceTest" "-Dsurefire.failIfNoSpecifiedTests=false" test`; expect the missing-method failure.
|
||
- [ ] Implement by reading `Set<String> commandIds = pendingIds().readAll()`, deleting every `pendingKey(commandId)`, clearing the set after iteration, and returning `commandIds.size()`. Do not call `deletePending` while iterating.
|
||
- [ ] Rerun the focused command; expect all `MqttCommandAckServiceTest` tests to pass.
|
||
- [ ] Commit only these two files with `git add water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java && git commit -m "fix: clear mqtt pending commands on request"`.
|
||
|
||
### Task 2: Clear Pending Commands After Application Startup
|
||
|
||
**Files:** create `water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttPendingCommandCleanupListener.java` and `water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttPendingCommandCleanupListenerTest.java`.
|
||
|
||
**Interface:** constructor consumes `MqttCommandAckService`; `onApplicationReady(ApplicationReadyEvent)` returns `void`.
|
||
|
||
- [ ] Write failing tests: verify a ready event calls `clearPendingCommands()`; stub it to throw `IllegalStateException("redis unavailable")` and assert the listener does not throw.
|
||
- [ ] Run `mvn -pl water-common/water-common-mqtt -am "-DskipTests=false" "-Dmaven.test.skip=false" "-Dtest=MqttPendingCommandCleanupListenerTest" "-Dsurefire.failIfNoSpecifiedTests=false" test`; expect the class-missing failure.
|
||
- [ ] Implement `@Slf4j @Component @RequiredArgsConstructor` with `@EventListener(ApplicationReadyEvent.class)`. Call the service, log the count, catch `RuntimeException`, and log `log.error("[MQTT] 应用启动清理待确认命令失败", e)` without rethrowing.
|
||
- [ ] Rerun the listener test; expect both tests to pass.
|
||
- [ ] Commit only the listener files with `git add water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttPendingCommandCleanupListener.java water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttPendingCommandCleanupListenerTest.java && git commit -m "fix: clear mqtt pending commands after startup"`.
|
||
|
||
### Task 3: Set New Retry Interval To Ten Seconds
|
||
|
||
**Files:** modify `water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/config/properties/MqttProperties.java`, `water-admin/src/main/resources/application.yml`; test `water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/DeviceMqttCommandPublisherTest.java`.
|
||
|
||
- [ ] Add a publisher timing assertion: after `send(command)`, assert `command.getNextRetryAt() - command.getLastSentAt()` equals `10000`.
|
||
- [ ] Run `mvn -pl water-common/water-common-mqtt -am "-DskipTests=false" "-Dmaven.test.skip=false" "-Dtest=DeviceMqttCommandPublisherTest" "-Dsurefire.failIfNoSpecifiedTests=false" test`; expect failure against the current 5000 millisecond default.
|
||
- [ ] Change `CommandAck.retryIntervalMs` default from `5000` to `10000`; change application YAML `retry-interval-ms` from `30000` to `10000`; leave scan interval `5000` and max retry count `3` unchanged.
|
||
- [ ] Run `mvn -pl water-admin -am "-DskipTests=false" "-Dmaven.test.skip=false" "-Dprofiles.active=dev" "-Dtest=DeviceMqttCommandPublisherTest,MqttCommandAckServiceTest,MqttPendingCommandCleanupListenerTest" "-Dsurefire.failIfNoSpecifiedTests=false" test`; expect `BUILD SUCCESS`.
|
||
- [ ] Commit configuration and publisher test with `git add water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/config/properties/MqttProperties.java water-admin/src/main/resources/application.yml water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/DeviceMqttCommandPublisherTest.java && git commit -m "fix: wait ten seconds before mqtt command retry"`.
|
||
|
||
### Task 4: Final Verification
|
||
|
||
- [ ] Run `mvn -pl water-common/water-common-mqtt -am "-DskipTests=false" "-Dmaven.test.skip=false" test`; expect `BUILD SUCCESS` and zero failures.
|
||
- [ ] Run `git diff --check` and `git status --short`; expect no whitespace errors and no reversion of unrelated worktree changes.
|
||
- [ ] Rebuild and restart the single application instance; confirm one startup cleanup log and that a newly sent command is retried no earlier than roughly 10 seconds later.
|