diff --git a/data/firmware/30d2bf300a59423583c406d7155fddd1.bin b/data/firmware/30d2bf300a59423583c406d7155fddd1.bin new file mode 100644 index 0000000..821c5e7 Binary files /dev/null and b/data/firmware/30d2bf300a59423583c406d7155fddd1.bin differ diff --git a/script/sql/update/add_device_binding_lifecycle.sql b/script/sql/update/add_device_binding_lifecycle.sql index 343050f..bae6e2b 100644 --- a/script/sql/update/add_device_binding_lifecycle.sql +++ b/script/sql/update/add_device_binding_lifecycle.sql @@ -10,7 +10,6 @@ CREATE TABLE app_device_binding ( device_name_snapshot varchar(255) NULL COMMENT '列表显示名称快照', unbound_time datetime NULL COMMENT '解绑完成时间', unbind_source varchar(16) NULL COMMENT 'DEVICE/APP/ADMIN', - unbind_command_id varchar(64) NULL COMMENT '触发解绑的命令编号', cleanup_attempts int NOT NULL DEFAULT 0 COMMENT '清理尝试次数', cleanup_error varchar(500) NULL COMMENT '最近清理错误', create_dept bigint NULL, @@ -32,8 +31,7 @@ CREATE TABLE app_device_binding ( CREATE TABLE app_device_unbind_command ( id bigint NOT NULL, device_no varchar(64) NOT NULL COMMENT '设备编号', - command_id varchar(64) NOT NULL COMMENT '设备解绑命令编号', - binding_id varchar(64) NOT NULL COMMENT '命令目标绑定标识', + binding_id varchar(64) NOT NULL COMMENT '解绑目标绑定标识', command_status varchar(16) NOT NULL COMMENT 'PROCESSING/APPLIED/STALE/FAILED', message varchar(500) NULL COMMENT '处理结果', attempt_count int NOT NULL DEFAULT 0 COMMENT '处理次数', @@ -44,9 +42,8 @@ CREATE TABLE app_device_unbind_command ( update_by bigint NULL, update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), - UNIQUE KEY uk_app_device_unbind_command (device_no, command_id), - KEY idx_app_device_unbind_binding (device_no, binding_id) -) COMMENT='设备解绑命令处理记录'; + UNIQUE KEY uk_app_device_unbind_binding (device_no, binding_id) +) COMMENT='设备解绑处理记录'; -- Existing bindings must re-establish the device-side bindingId before becoming ACTIVE. INSERT INTO app_device_binding ( diff --git a/script/sql/update/remove_device_unbind_command_id.sql b/script/sql/update/remove_device_unbind_command_id.sql new file mode 100644 index 0000000..ff342b8 --- /dev/null +++ b/script/sql/update/remove_device_unbind_command_id.sql @@ -0,0 +1,35 @@ +-- Device-initiated unbinding is identified by device_no + binding_id. +-- Keep the most useful result if the old commandId protocol created duplicate rows. +DELETE d +FROM app_device_unbind_command d +JOIN app_device_unbind_command r + ON r.device_no = d.device_no + AND r.binding_id = d.binding_id + AND ( + CASE r.command_status + WHEN 'APPLIED' THEN 4 + WHEN 'PROCESSING' THEN 3 + WHEN 'FAILED' THEN 2 + WHEN 'STALE' THEN 1 + ELSE 0 + END > CASE d.command_status + WHEN 'APPLIED' THEN 4 + WHEN 'PROCESSING' THEN 3 + WHEN 'FAILED' THEN 2 + WHEN 'STALE' THEN 1 + ELSE 0 + END + OR ( + r.command_status = d.command_status + AND r.id < d.id + ) + ); + +ALTER TABLE app_device_binding + DROP COLUMN unbind_command_id; + +ALTER TABLE app_device_unbind_command + DROP INDEX uk_app_device_unbind_command, + DROP INDEX idx_app_device_unbind_binding, + DROP COLUMN command_id, + ADD UNIQUE KEY uk_app_device_unbind_binding (device_no, binding_id); 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 8beb17f..ec8b06e 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 @@ -365,6 +365,11 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler { // 其他线程正在刷新，本次跳过；状态写入幂等，下条 ACK 仍会兜底刷新 return; } + appDeviceMapper.update(null, + new LambdaUpdateWrapper() + .set(AppDevice::getStatus, "1") + .eq(AppDevice::getDeviceNo, deviceNo) + ); Map statusCache = new HashMap<>(); statusCache.put("deviceNo", deviceNo); statusCache.put("status", "1"); @@ -374,11 +379,6 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler { statusCache, Duration.ofSeconds(mqttProperties.getCommandAck().getDeviceStatusCacheTtlSeconds()) ); - appDeviceMapper.update(null, - new LambdaUpdateWrapper() - .set(AppDevice::getStatus, "1") - .eq(AppDevice::getDeviceNo, deviceNo) - ); } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn("[MQTT] 刷新设备在线状态被中断 设备编号={}", deviceNo); diff --git a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisher.java b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisher.java index 50c4f58..dff322b 100644 --- a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisher.java +++ b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisher.java @@ -30,7 +30,6 @@ public class MqttDeviceUnbindResultPublisher implements IDeviceUnbindResultPubli throw new ServiceException("解绑回执缺少设备编号"); } Map payload = new LinkedHashMap<>(); - payload.put("commandId", result.commandId()); payload.put("bindingId", result.bindingId()); payload.put("status", result.status()); payload.put("message", result.message()); 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 b2c800b..6d70fbf 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 @@ -418,10 +418,23 @@ class MqttCommandAckServiceTest { when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); when(lock.isHeldByCurrentThread()).thenReturn(true); + AtomicBoolean databaseUpdated = new AtomicBoolean(); + doAnswer(invocation -> { + databaseUpdated.set(true); + return 1; + }).when(appDeviceMapper).update(eq(null), any()); MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper); try (MockedStatic redis = mockStatic(RedisUtils.class)) { redis.when(RedisUtils::getClient).thenReturn(redissonClient); + redis.when(() -> RedisUtils.setCacheObject( + eq("mqtt:device:status:D01"), any(Map.class), eq(Duration.ofSeconds(900)) + )).thenAnswer(invocation -> { + if (!databaseUpdated.get()) { + throw new AssertionError("online cache was written before database status update"); + } + return null; + }); ReflectionTestUtils.invokeMethod(service, "refreshDeviceOnline", "D01"); @@ -434,4 +447,30 @@ class MqttCommandAckServiceTest { verify(lock).unlock(); } } + + @Test + void refreshDeviceOnlineDoesNotWriteCacheWhenDatabaseUpdateFails() throws Exception { + MqttProperties properties = new MqttProperties(); + AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class); + RedissonClient redissonClient = mock(RedissonClient.class); + RLock lock = mock(RLock.class); + when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); + when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); + when(lock.isHeldByCurrentThread()).thenReturn(true); + doThrow(new IllegalStateException("database failure")).when(appDeviceMapper).update(eq(null), any()); + + MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper); + try (MockedStatic redis = mockStatic(RedisUtils.class)) { + redis.when(RedisUtils::getClient).thenReturn(redissonClient); + + assertThatThrownBy(() -> ReflectionTestUtils.invokeMethod(service, "refreshDeviceOnline", "D01")) + .isInstanceOf(IllegalStateException.class) + .hasMessage("database failure"); + + redis.verify(() -> RedisUtils.setCacheObject( + eq("mqtt:device:status:D01"), any(Map.class), eq(Duration.ofSeconds(600)) + ), never()); + verify(lock).unlock(); + } + } } diff --git a/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisherTest.java b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisherTest.java new file mode 100644 index 0000000..57d90b4 --- /dev/null +++ b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttDeviceUnbindResultPublisherTest.java @@ -0,0 +1,58 @@ +package org.dromara.mqtt; + +import cn.hutool.extra.spring.SpringUtil; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.dromara.app.domain.mqtt.DeviceUnbindResult; +import org.dromara.mqtt.config.properties.MqttProperties; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.context.support.GenericApplicationContext; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.function.Supplier; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +@Tag("dev") +class MqttDeviceUnbindResultPublisherTest { + + private static GenericApplicationContext applicationContext; + + @BeforeAll + static void initializeJsonUtils() { + applicationContext = new GenericApplicationContext(); + applicationContext.registerBean(ObjectMapper.class, (Supplier) ObjectMapper::new); + applicationContext.refresh(); + new SpringUtil().setApplicationContext(applicationContext); + } + + @AfterAll + static void closeApplicationContext() { + applicationContext.close(); + } + + @Test + void publishUsesBindingIdWithoutCommandId() throws Exception { + MqttClientManager clientManager = mock(MqttClientManager.class); + MqttDeviceUnbindResultPublisher publisher = new MqttDeviceUnbindResultPublisher(new MqttProperties()); + ReflectionTestUtils.setField(publisher, "mqttClientManager", clientManager); + + publisher.publish(new DeviceUnbindResult( + "D01", "binding-1", DeviceUnbindResult.APPLIED, "done")); + + ArgumentCaptor payloadCaptor = ArgumentCaptor.forClass(String.class); + verify(clientManager).publish(org.mockito.ArgumentMatchers.eq("/d01/subscriber/unbind/ack"), + payloadCaptor.capture()); + JsonNode payload = new ObjectMapper().readTree(payloadCaptor.getValue()); + assertThat(payload.has("commandId")).isFalse(); + assertThat(payload.get("bindingId").asText()).isEqualTo("binding-1"); + assertThat(payload.get("status").asText()).isEqualTo(DeviceUnbindResult.APPLIED); + assertThat(payload.get("message").asText()).isEqualTo("done"); + } +} diff --git a/water-common/water-common-mybatis/pom.xml b/water-common/water-common-mybatis/pom.xml index ee7a50d..2849920 100644 --- a/water-common/water-common-mybatis/pom.xml +++ b/water-common/water-common-mybatis/pom.xml @@ -47,6 +47,12 @@ p6spy p6spy + + + org.springframework.boot + spring-boot-starter-test + test + diff --git a/water-common/water-common-mybatis/src/main/java/org/dromara/common/mybatis/p6spy/ScheduledTaskAwareStdoutLogger.java b/water-common/water-common-mybatis/src/main/java/org/dromara/common/mybatis/p6spy/ScheduledTaskAwareStdoutLogger.java new file mode 100644 index 0000000..9078835 --- /dev/null +++ b/water-common/water-common-mybatis/src/main/java/org/dromara/common/mybatis/p6spy/ScheduledTaskAwareStdoutLogger.java @@ -0,0 +1,27 @@ +package org.dromara.common.mybatis.p6spy; + +import com.baomidou.mybatisplus.extension.p6spy.StdoutLogger; + +/** + * P6Spy SQL logger that suppresses SQL executed by scheduled tasks. + */ +public class ScheduledTaskAwareStdoutLogger extends StdoutLogger { + + private static final String SCHEDULE_POOL_PREFIX = "schedule-pool-"; + private static final String VIRTUAL_SCHEDULE_POOL_PREFIX = "virtual-schedule-pool-"; + private static final String SPRING_SCHEDULING_PREFIX = "scheduling-"; + + static boolean isScheduledTaskThread(String threadName) { + return threadName != null + && (threadName.startsWith(SCHEDULE_POOL_PREFIX) + || threadName.startsWith(VIRTUAL_SCHEDULE_POOL_PREFIX) + || threadName.startsWith(SPRING_SCHEDULING_PREFIX)); + } + + @Override + public void logText(String text) { + if (!isScheduledTaskThread(Thread.currentThread().getName())) { + super.logText(text); + } + } +} diff --git a/water-common/water-common-mybatis/src/main/resources/spy.properties b/water-common/water-common-mybatis/src/main/resources/spy.properties index f3ed7d8..24ea8f4 100644 --- a/water-common/water-common-mybatis/src/main/resources/spy.properties +++ b/water-common/water-common-mybatis/src/main/resources/spy.properties @@ -2,8 +2,8 @@ modulelist=com.baomidou.mybatisplus.extension.p6spy.MybatisPlusLogFactory,com.p6spy.engine.outage.P6OutageFactory # 自定义日志打印 logMessageFormat=com.baomidou.mybatisplus.extension.p6spy.P6SpyLogger -#日志输出到控制台 -appender=com.baomidou.mybatisplus.extension.p6spy.StdoutLogger +# 日志输出到控制台，定时任务线程不打印 SQL +appender=org.dromara.common.mybatis.p6spy.ScheduledTaskAwareStdoutLogger # 使用日志系统记录 sql #appender=com.p6spy.engine.spy.appender.Slf4JLogger # 取消JDBC URL前缀 diff --git a/water-common/water-common-mybatis/src/test/java/org/dromara/common/mybatis/p6spy/ScheduledTaskAwareStdoutLoggerTest.java b/water-common/water-common-mybatis/src/test/java/org/dromara/common/mybatis/p6spy/ScheduledTaskAwareStdoutLoggerTest.java new file mode 100644 index 0000000..c3c438f --- /dev/null +++ b/water-common/water-common-mybatis/src/test/java/org/dromara/common/mybatis/p6spy/ScheduledTaskAwareStdoutLoggerTest.java @@ -0,0 +1,24 @@ +package org.dromara.common.mybatis.p6spy; + +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +@Tag("dev") +public class ScheduledTaskAwareStdoutLoggerTest { + + @Test + public void identifiesScheduledTaskThreads() { + assertThat(ScheduledTaskAwareStdoutLogger.isScheduledTaskThread("schedule-pool-1")).isTrue(); + assertThat(ScheduledTaskAwareStdoutLogger.isScheduledTaskThread("virtual-schedule-pool-2")).isTrue(); + assertThat(ScheduledTaskAwareStdoutLogger.isScheduledTaskThread("scheduling-3")).isTrue(); + } + + @Test + public void keepsSqlLoggingEnabledForOtherThreads() { + assertThat(ScheduledTaskAwareStdoutLogger.isScheduledTaskThread("XNIO-1 task-2")).isFalse(); + assertThat(ScheduledTaskAwareStdoutLogger.isScheduledTaskThread("mqtt-consumer-1")).isFalse(); + assertThat(ScheduledTaskAwareStdoutLogger.isScheduledTaskThread(null)).isFalse(); + } +} diff --git a/water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java b/water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java index a5b8c96..1a853f4 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java @@ -130,7 +130,9 @@ public class AppController extends BaseController { @GetMapping("/device/{deviceNo}") public R getDeviceInfo(@NotNull(message = "主键不能为空") @PathVariable String deviceNo) { - return R.ok(getActiveDevice(deviceNo)); + AppDeviceVo device = getActiveDevice(deviceNo); + deviceCommandService.sendQueryPowerCommand(deviceNo, LoginHelper.getUserId()); + return R.ok(device); } @@ -778,7 +780,7 @@ public class AppController extends BaseController { Long userId = LoginHelper.getUserId(); AppDeviceVo device = appDeviceService.queryActiveDevice(userId, deviceNo); if (device == null) { - throw new ServiceException("app.device.not.exists.or.denied"); + throw new ServiceException("app.device.not.exists.or.denied", 501); } return device; } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceBinding.java b/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceBinding.java index 32dc02f..0968e3d 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceBinding.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceBinding.java @@ -28,7 +28,6 @@ public class AppDeviceBinding extends BaseEntity { private String deviceNameSnapshot; private Date unboundTime; private String unbindSource; - private String unbindCommandId; private Integer cleanupAttempts; private String cleanupError; } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceUnbindCommand.java b/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceUnbindCommand.java index db1cd8b..80dcf0d 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceUnbindCommand.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/domain/AppDeviceUnbindCommand.java @@ -20,7 +20,6 @@ public class AppDeviceUnbindCommand extends BaseEntity { @TableId("id") private Long id; private String deviceNo; - private String commandId; private String bindingId; private String commandStatus; private String message; diff --git a/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindRequest.java b/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindRequest.java index bb50374..d4cc845 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindRequest.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindRequest.java @@ -4,7 +4,6 @@ import lombok.Data; @Data public class DeviceUnbindRequest { - private String commandId; private String bindingId; private String reason; } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindResult.java b/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindResult.java index b97f53c..f4ccb61 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindResult.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceUnbindResult.java @@ -2,7 +2,6 @@ package org.dromara.app.domain.mqtt; public record DeviceUnbindResult( String deviceNo, - String commandId, String bindingId, String status, String message diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java index 6ce41b4..38e0f41 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java @@ -7,6 +7,7 @@ import org.dromara.app.domain.bo.AppDeviceBo; import org.dromara.app.mqtt.MqttTopicHandler; import org.dromara.app.service.IAppDeviceBindingService; import org.dromara.app.service.IAppDeviceService; +import org.dromara.common.core.utils.StringUtils; import org.dromara.common.json.utils.JsonUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; @@ -49,7 +50,7 @@ public class DeviceDataHandler implements MqttTopicHandler { return; } Map dto = JsonUtils.parseObject(payload, Map.class); - if (!MqttBindingReportGuard.accepts(deviceBindingService, deviceNo, dto)) { + if (!acceptsPowerReport(deviceNo, dto)) { log.debug("[MQTT] 忽略非当前绑定设备电量上报 设备编号={}", deviceNo); return; } @@ -73,4 +74,17 @@ public class DeviceDataHandler implements MqttTopicHandler { } } + /** + * 电量报文兼容未携带 bindingId 的旧设备，但仍只接受 ACTIVE 绑定设备的上报。 + */ + private boolean acceptsPowerReport(String deviceNo, Map payload) { + if (!deviceBindingService.isActive(deviceNo)) { + return false; + } + String bindingId = payload == null || payload.get("bindingId") == null + ? null : String.valueOf(payload.get("bindingId")); + return StringUtils.isBlank(bindingId) + || deviceBindingService.isCurrentBinding(deviceNo, bindingId); + } + } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java index 5975e13..df956a0 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceStatusHandler.java @@ -61,8 +61,14 @@ public class DeviceStatusHandler implements MqttTopicHandler { log.warn("[MQTT] 忽略 retained 离线状态 时间={} 设备编号={} 消息体={}", HandlerLogTime.now(), deviceNo, payload); return; } - deviceStatusService.markOffline(deviceNo, OFFLINE_REASON); - log.info("[MQTT] 设备状态离线 时间={} 设备编号={} 消息体={}", HandlerLogTime.now(), deviceNo, payload); + boolean reconciled = deviceStatusService.markOfflineIfCacheMissing(deviceNo, OFFLINE_REASON); + if (reconciled) { + log.info("[MQTT] 设备在线缓存缺失，已协调离线 时间={} 设备编号={} 消息体={}", + HandlerLogTime.now(), deviceNo, payload); + } else { + log.debug("[MQTT] 设备离线状态未变更，可能仍有在线缓存、已离线或状态锁不可用 时间={} 设备编号={} 消息体={}", + HandlerLogTime.now(), deviceNo, payload); + } return; } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceUnbindHandler.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceUnbindHandler.java index a3f2888..54bfef4 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceUnbindHandler.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceUnbindHandler.java @@ -42,19 +42,19 @@ public class DeviceUnbindHandler implements MqttTopicHandler { } DeviceUnbindResult result = bindingService.unbindByDevice(deviceNo, request); resultPublisher.publish(result); - log.info("[MQTT] 设备解绑已处理 设备编号={} 命令编号={} 结果={}", - deviceNo, result.commandId(), result.status()); + log.info("[MQTT] 设备解绑已处理 设备编号={} 绑定标识={} 结果={}", + deviceNo, result.bindingId(), result.status()); } catch (RuntimeException e) { log.error("[MQTT] 设备解绑处理失败 设备标识={} 设备编号={} 消息体={}", deviceIdentity, deviceNo, payload, e); - if (deviceNo != null && request != null && request.getCommandId() != null) { + if (deviceNo != null && request != null) { try { resultPublisher.publish(new DeviceUnbindResult( - deviceNo, request.getCommandId(), request.getBindingId(), + deviceNo, request.getBindingId(), DeviceUnbindResult.FAILED, e.getMessage())); } catch (RuntimeException publishFailure) { - log.warn("[MQTT] 设备解绑失败回执发送失败 设备编号={} 命令编号={}", - deviceNo, request.getCommandId(), publishFailure); + log.warn("[MQTT] 设备解绑失败回执发送失败 设备编号={} 绑定标识={}", + deviceNo, request.getBindingId(), publishFailure); } } } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java index 3c642a0..0bdd70c 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/handler/MqttDeviceStatusService.java @@ -6,8 +6,8 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.dromara.app.domain.AppDevice; import org.dromara.app.mapper.AppDeviceMapper; +import org.dromara.app.service.DeviceOfflineReconcileService; import org.dromara.app.service.IAppDeviceBindingService; -import org.dromara.app.service.IAppWateringLogService; import org.dromara.common.core.utils.StringUtils; import org.dromara.common.redis.utils.RedisUtils; import org.redisson.api.RLock; @@ -31,7 +31,7 @@ public class MqttDeviceStatusService { private static final String DEVICE_STATUS_LOCK_PREFIX = "lock:mqtt:device:status:"; private final AppDeviceMapper appDeviceMapper; - private final IAppWateringLogService wateringLogService; + private final DeviceOfflineReconcileService deviceOfflineReconcileService; @Lazy @Autowired private IAppDeviceBindingService deviceBindingService; @@ -60,6 +60,11 @@ public class MqttDeviceStatusService { private void markOnlineInternal(String deviceNo) { withDeviceStatusLock(deviceNo, () -> { + appDeviceMapper.update(null, + new LambdaUpdateWrapper() + .set(AppDevice::getStatus, "1") + .eq(AppDevice::getDeviceNo, deviceNo) + ); Map statusCache = new HashMap<>(); statusCache.put("deviceNo", deviceNo); statusCache.put("status", "1"); @@ -69,11 +74,6 @@ public class MqttDeviceStatusService { statusCache, Duration.ofSeconds(deviceStatusCacheTtlSeconds) ); - appDeviceMapper.update(null, - new LambdaUpdateWrapper() - .set(AppDevice::getStatus, "1") - .eq(AppDevice::getDeviceNo, deviceNo) - ); return true; }); } @@ -95,28 +95,13 @@ public class MqttDeviceStatusService { } private boolean markOfflineWithLock(String deviceNo, String reason, boolean skipWhenCacheExists) { - if (!isActive(deviceNo)) { - return false; - } return withDeviceStatusLock(deviceNo, () -> { if (skipWhenCacheExists && RedisUtils.hasKey(deviceStatusCachePrefix + deviceNo)) { return false; } + boolean reconciled = deviceOfflineReconcileService.reconcileOffline(deviceNo, reason, new Date()); RedisUtils.deleteObject(deviceStatusCachePrefix + deviceNo); - int updated = appDeviceMapper.update(null, - new LambdaUpdateWrapper() - .set(AppDevice::getStatus, "0") - .set(AppDevice::getWorkStatus, "0") - .ne(AppDevice::getStatus, "0") - .eq(AppDevice::getDeviceNo, deviceNo) - ); - if (updated > 0) { - int count = wateringLogService.finishRunningLogsByDevice(deviceNo, new Date(), reason); - if (count > 0) { - log.info("[MQTT] 已结束设备进行中的浇水记录 设备编号={} 记录数={} 原因={}", deviceNo, count, reason); - } - } - return updated > 0; + return reconciled; }); } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceMapper.java b/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceMapper.java index 1daae8c..ebb9403 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceMapper.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceMapper.java @@ -238,6 +238,33 @@ public interface AppDeviceMapper extends BaseMapperPlus """) AppDevice selectByMac(@Param("macAddress") String macAddress); + @Update(""" + update app_device + set status = '0', + work_status = '0' + where device_no = #{deviceNo} + and ( + coalesce(status, '') <> '0' + or coalesce(work_status, '') <> '0' + ) + """) + int markOfflineAndStopWork(@Param("deviceNo") String deviceNo); + + @Select(""" + select d.device_no + from app_device d + where coalesce(d.status, '') <> '0' + or coalesce(d.work_status, '') <> '0' + or exists ( + select 1 + from app_watering_log l + where l.device_no = d.device_no + and l.status = '1' + ) + order by d.device_no asc + """) + List selectOfflineReconcileCandidateDeviceNos(); + @Select(""" select * from app_device diff --git a/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceUnbindCommandMapper.java b/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceUnbindCommandMapper.java index a952f9c..70e3055 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceUnbindCommandMapper.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppDeviceUnbindCommandMapper.java @@ -10,9 +10,9 @@ public interface AppDeviceUnbindCommandMapper @Select(""" select * from app_device_unbind_command - where device_no = #{deviceNo} and command_id = #{commandId} + where device_no = #{deviceNo} and binding_id = #{bindingId} limit 1 """) - AppDeviceUnbindCommand selectByDeviceAndCommandId(@Param("deviceNo") String deviceNo, - @Param("commandId") String commandId); + AppDeviceUnbindCommand selectByDeviceAndBindingId(@Param("deviceNo") String deviceNo, + @Param("bindingId") String bindingId); } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppWateringLogMapper.java b/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppWateringLogMapper.java index 10bdd75..2cbbb69 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppWateringLogMapper.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/mapper/AppWateringLogMapper.java @@ -2,10 +2,13 @@ package org.dromara.app.mapper; import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.annotations.Update; import org.dromara.app.domain.AppWateringLog; import org.dromara.app.domain.vo.AppWateringLogVo; import org.dromara.common.mybatis.core.mapper.BaseMapperPlus; +import java.util.Date; + /** * 浇水记录Mapper接口 * @@ -52,4 +55,17 @@ public interface AppWateringLogMapper extends BaseMapperPlus '' then #{remark} else remark end + where device_no = #{deviceNo} + and status = '1' + """) + int finishRunningLogsByDevice(@Param("deviceNo") String deviceNo, + @Param("endTime") Date endTime, + @Param("remark") String remark); } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/DeviceOfflineReconcileService.java b/water-modules/water-app/src/main/java/org/dromara/app/service/DeviceOfflineReconcileService.java new file mode 100644 index 0000000..cb096db --- /dev/null +++ b/water-modules/water-app/src/main/java/org/dromara/app/service/DeviceOfflineReconcileService.java @@ -0,0 +1,26 @@ +package org.dromara.app.service; + +import lombok.RequiredArgsConstructor; +import org.dromara.app.mapper.AppDeviceMapper; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.util.Date; + +/** + * Coordinates the database changes required when a device becomes offline. + */ +@Service +@RequiredArgsConstructor +public class DeviceOfflineReconcileService { + + private final AppDeviceMapper appDeviceMapper; + private final IAppWateringLogService wateringLogService; + + @Transactional(rollbackFor = Exception.class) + public boolean reconcileOffline(String deviceNo, String reason, Date offlineTime) { + int deviceUpdated = appDeviceMapper.markOfflineAndStopWork(deviceNo); + int finishedLogs = wateringLogService.finishRunningLogsByDevice(deviceNo, offlineTime, reason); + return deviceUpdated > 0 || finishedLogs > 0; + } +} \ No newline at end of file diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceBindingServiceImpl.java b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceBindingServiceImpl.java index 29e7447..a16e455 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceBindingServiceImpl.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceBindingServiceImpl.java @@ -257,8 +257,8 @@ public class AppDeviceBindingServiceImpl @Lock4j(keys = {"'device:command:' + #deviceNo"}, acquireTimeout = 5000, expire = 30000) public DeviceUnbindResult unbindByDevice(String deviceNo, DeviceUnbindRequest request) { validateDeviceUnbindRequest(deviceNo, request); - AppDeviceUnbindCommand existing = unbindCommandMapper.selectByDeviceAndCommandId( - deviceNo, request.getCommandId()); + AppDeviceUnbindCommand existing = unbindCommandMapper.selectByDeviceAndBindingId( + deviceNo, request.getBindingId()); if (existing != null) { if (DeviceUnbindCommandStatus.APPLIED.equals(existing.getCommandStatus())) { return result(existing, DeviceUnbindResult.ALREADY_APPLIED, "解绑已完成"); @@ -272,22 +272,22 @@ public class AppDeviceBindingServiceImpl ? transactionTemplate.execute(status -> acceptDeviceUnbind(deviceNo, request)) : resumeDeviceUnbind(existing); if (acceptance == null) { - return new DeviceUnbindResult(deviceNo, request.getCommandId(), request.getBindingId(), + return new DeviceUnbindResult(deviceNo, request.getBindingId(), DeviceUnbindResult.FAILED, "解绑处理失败"); } if (acceptance.stale()) { - return new DeviceUnbindResult(deviceNo, request.getCommandId(), request.getBindingId(), + return new DeviceUnbindResult(deviceNo, request.getBindingId(), DeviceUnbindResult.STALE_BINDING, "绑定标识已过期"); } try { clearRuntimeAfterAccepted(deviceNo); AppDeviceBinding completed = cleanupBinding(acceptance.binding().getId()); - return new DeviceUnbindResult(deviceNo, request.getCommandId(), completed.getBindingId(), + return new DeviceUnbindResult(deviceNo, completed.getBindingId(), DeviceUnbindResult.APPLIED, "解绑完成"); } catch (RuntimeException e) { markCleanupFailed(acceptance.binding(), e); - return new DeviceUnbindResult(deviceNo, request.getCommandId(), request.getBindingId(), + return new DeviceUnbindResult(deviceNo, request.getBindingId(), DeviceUnbindResult.FAILED, "解绑清理失败，服务器将继续重试"); } } @@ -442,7 +442,6 @@ public class AppDeviceBindingServiceImpl relationship.setDeviceNameSnapshot(deviceName); relationship.setUnboundTime(null); relationship.setUnbindSource(null); - relationship.setUnbindCommandId(null); relationship.setCleanupAttempts(0); relationship.setCleanupError(null); bindingMapper.updateById(relationship); @@ -495,7 +494,6 @@ public class AppDeviceBindingServiceImpl .set(AppDeviceBinding::getDeviceNameSnapshot, device == null ? binding.getDeviceNameSnapshot() : device.getDeviceName()) .set(AppDeviceBinding::getUnbindSource, source) - .set(AppDeviceBinding::getUnbindCommandId, UUID.randomUUID().toString().replace("-", "")) .set(AppDeviceBinding::getCleanupAttempts, 0) .set(AppDeviceBinding::getCleanupError, null) .eq(AppDeviceBinding::getId, binding.getId()) @@ -523,7 +521,6 @@ public class AppDeviceBindingServiceImpl .set(AppDeviceBinding::getDeviceNameSnapshot, device == null ? binding.getDeviceNameSnapshot() : device.getDeviceName()) .set(AppDeviceBinding::getUnbindSource, DeviceUnbindSource.DEVICE) - .set(AppDeviceBinding::getUnbindCommandId, request.getCommandId()) .set(AppDeviceBinding::getCleanupAttempts, 0) .set(AppDeviceBinding::getCleanupError, null) .eq(AppDeviceBinding::getId, binding.getId()) @@ -571,7 +568,7 @@ public class AppDeviceBindingServiceImpl .set(AppDeviceUnbindCommand::getCompletedTime, new Date()) .setSql("attempt_count = attempt_count + 1") .eq(AppDeviceUnbindCommand::getDeviceNo, binding.getDeviceNo()) - .eq(AppDeviceUnbindCommand::getCommandId, binding.getUnbindCommandId())); + .eq(AppDeviceUnbindCommand::getBindingId, binding.getBindingId())); } return bindingMapper.selectById(binding.getId()); }); @@ -609,7 +606,7 @@ public class AppDeviceBindingServiceImpl .set(AppDeviceUnbindCommand::getMessage, message) .setSql("attempt_count = attempt_count + 1") .eq(AppDeviceUnbindCommand::getDeviceNo, binding.getDeviceNo()) - .eq(AppDeviceUnbindCommand::getCommandId, binding.getUnbindCommandId())); + .eq(AppDeviceUnbindCommand::getBindingId, binding.getBindingId())); } }); AppDeviceBinding current = bindingMapper.selectById(binding.getId()); @@ -678,18 +675,17 @@ public class AppDeviceBindingServiceImpl } try { unbindResultPublisher.publish(new DeviceUnbindResult( - binding.getDeviceNo(), binding.getUnbindCommandId(), binding.getBindingId(), + binding.getDeviceNo(), binding.getBindingId(), DeviceUnbindResult.APPLIED, "解绑完成")); } catch (RuntimeException e) { - log.warn("[设备解绑] 恢复完成但 MQTT 回执发送失败 设备编号={} 命令编号={}", - binding.getDeviceNo(), binding.getUnbindCommandId(), e); + log.warn("[设备解绑] 恢复完成但 MQTT 回执发送失败 设备编号={} 绑定标识={}", + binding.getDeviceNo(), binding.getBindingId(), e); } } private void saveUnbindCommand(String deviceNo, DeviceUnbindRequest request, String status, String message) { AppDeviceUnbindCommand command = new AppDeviceUnbindCommand(); command.setDeviceNo(deviceNo); - command.setCommandId(request.getCommandId()); command.setBindingId(request.getBindingId()); command.setCommandStatus(status); command.setMessage(message); @@ -700,7 +696,7 @@ public class AppDeviceBindingServiceImpl try { unbindCommandMapper.insert(command); } catch (DuplicateKeyException e) { - log.debug("[设备解绑] 命令已存在 设备编号={} 命令编号={}", deviceNo, request.getCommandId()); + log.debug("[设备解绑] 处理记录已存在 设备编号={} 绑定标识={}", deviceNo, request.getBindingId()); } } @@ -769,8 +765,7 @@ public class AppDeviceBindingServiceImpl } private DeviceUnbindResult result(AppDeviceUnbindCommand command, String status, String message) { - return new DeviceUnbindResult(command.getDeviceNo(), command.getCommandId(), - command.getBindingId(), status, message); + return new DeviceUnbindResult(command.getDeviceNo(), command.getBindingId(), status, message); } private String abbreviate(String message, int maxLength) { diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppWateringLogServiceImpl.java b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppWateringLogServiceImpl.java index ba206ec..4126b2a 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppWateringLogServiceImpl.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppWateringLogServiceImpl.java @@ -285,31 +285,7 @@ public class AppWateringLogServiceImpl implements IAppWateringLogService { if (StringUtils.isBlank(deviceNo) || endTime == null) { return 0; } - - AppWateringLog activeLog = baseMapper.selectOne( - Wrappers.lambdaQuery() - .eq(AppWateringLog::getDeviceNo, deviceNo) - .eq(AppWateringLog::getStatus, "1") - .orderByDesc(AppWateringLog::getStartTime) - .orderByDesc(AppWateringLog::getCreateTime) - .orderByDesc(AppWateringLog::getId) - .last("limit 1") - ); - if (activeLog == null) { - return 0; - } - - LambdaUpdateWrapper updateWrapper = Wrappers.lambdaUpdate(); - updateWrapper.eq(AppWateringLog::getId, activeLog.getId()) - .set(AppWateringLog::getEndTime, endTime) - .set(AppWateringLog::getStatus, "0") - .set(StringUtils.isNotBlank(remark), AppWateringLog::getRemark, remark); - if (activeLog.getStartTime() != null) { - long durationMillis = endTime.getTime() - activeLog.getStartTime().getTime(); - long durationMin = Math.max(durationMillis, 0L) / 60000L; - updateWrapper.set(AppWateringLog::getDurationMin, String.valueOf(durationMin)); - } - return baseMapper.update(null, updateWrapper); + return baseMapper.finishRunningLogsByDevice(deviceNo, endTime, remark); } /** diff --git a/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java b/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java index 142d41b..cf8fa5e 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java +++ b/water-modules/water-app/src/main/java/org/dromara/app/task/DeviceOfflineCheckTask.java @@ -1,9 +1,7 @@ package org.dromara.app.task; -import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.dromara.app.domain.AppDevice; import org.dromara.app.handler.MqttDeviceStatusService; import org.dromara.app.mapper.AppDeviceMapper; import org.dromara.common.redis.utils.RedisUtils; @@ -12,7 +10,6 @@ import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.time.Duration; -import java.util.ArrayList; import java.util.List; /** @@ -20,8 +17,8 @@ import java.util.List; *

* 逻辑：电量上报刷新 Redis 在线 Key，Key 超时后判定设备离线。 * 设备 LWT 离线消息仍可立即更新离线状态，本任务用于处理未收到遗嘱的异常断线。 - * 本任务定期扫描数据库中非离线状态的设备，如果对应 Redis Key 已不存在， - * 则将数据库 status 更新为 0（离线）。 + * 本任务定期扫描未离线、未停止工作或存在运行日志的设备，如果对应 Redis Key 已不存在， + * 则协调数据库中的离线状态、工作状态和运行日志。 *

* *

扫描间隔通过 {@code mqtt.offline-check.interval-ms} 配置，建议小于设备在线 Key 的 TTL。

@@ -55,33 +52,35 @@ public class DeviceOfflineCheckTask { return; } - // 查询数据库中所有非离线状态的设备，避免状态不是 1 但仍需离线修正的设备被漏掉 - List onlineDevices = appDeviceMapper.selectList( - new LambdaQueryWrapper() - .ne(AppDevice::getStatus, "0") - .select(AppDevice::getDeviceNo) - ); - - if (onlineDevices.isEmpty()) { + List candidateDeviceNos = appDeviceMapper.selectOfflineReconcileCandidateDeviceNos(); + if (candidateDeviceNos.isEmpty()) { return; } - List offlineDeviceNos = new ArrayList<>(); + List updatedDeviceNos = new java.util.ArrayList<>(); int migratedPermanentKeys = 0; - for (AppDevice device : onlineDevices) { - String deviceNo = device.getDeviceNo(); - String statusKey = deviceStatusCachePrefix + deviceNo; - if (!RedisUtils.hasKey(statusKey)) { - offlineDeviceNos.add(deviceNo); + for (String deviceNo : candidateDeviceNos) { + if (deviceNo == null || deviceNo.isBlank()) { continue; } - - long ttl = RedisUtils.getTimeToLive(statusKey); - if (ttl == -1L) { - RedisUtils.expire(statusKey, Duration.ofSeconds(deviceStatusCacheTtlSeconds)); - migratedPermanentKeys++; - } else if (ttl == -2L) { - offlineDeviceNos.add(deviceNo); + try { + String statusKey = deviceStatusCachePrefix + deviceNo; + if (RedisUtils.hasKey(statusKey)) { + long ttl = RedisUtils.getTimeToLive(statusKey); + if (ttl == -1L) { + RedisUtils.expire(statusKey, Duration.ofSeconds(deviceStatusCacheTtlSeconds)); + migratedPermanentKeys++; + continue; + } + if (ttl != -2L) { + continue; + } + } + if (deviceStatusService.markOfflineIfCacheMissing(deviceNo, "设备电量心跳超时")) { + updatedDeviceNos.add(deviceNo); + } + } catch (Exception e) { + log.warn("[设备离线] 处理设备离线状态失败，设备编号={}", deviceNo, e); } } @@ -89,18 +88,6 @@ public class DeviceOfflineCheckTask { log.info("[设备离线] 已为旧版永久在线缓存补充 TTL，设备数：{}", migratedPermanentKeys); } - if (offlineDeviceNos.isEmpty()) { - log.debug("[设备离线] 本次检测未发现离线设备，扫描设备数：{}", onlineDevices.size()); - return; - } - - List updatedDeviceNos = new ArrayList<>(); - for (String deviceNo : offlineDeviceNos) { - if (deviceStatusService.markOfflineIfCacheMissing(deviceNo, "设备电量心跳超时")) { - updatedDeviceNos.add(deviceNo); - } - } - if (!updatedDeviceNos.isEmpty()) { log.info("[设备离线] 检测到 {} 台设备离线，已更新数据库状态，设备编号：{}", updatedDeviceNos.size(), updatedDeviceNos); diff --git a/water-modules/water-app/src/test/java/org/dromara/app/controller/AppControllerTest.java b/water-modules/water-app/src/test/java/org/dromara/app/controller/AppControllerTest.java index f692f79..df64b44 100644 --- a/water-modules/water-app/src/test/java/org/dromara/app/controller/AppControllerTest.java +++ b/water-modules/water-app/src/test/java/org/dromara/app/controller/AppControllerTest.java @@ -280,6 +280,34 @@ public class AppControllerTest { assertThatThrownBy(() -> callWithLogin(1L, () -> controller.getDeviceInfo("D01"))) .isSameAs(failure); + verifyNoInteractions(deviceCommandService); + } + + @Test + public void getDeviceInfo_uses501WhenDeviceDoesNotExistOrAccessIsDenied() { + AppController controller = newController(); + when(appDeviceService.queryActiveDevice(1L, "D01")).thenReturn(null); + + assertThatThrownBy(() -> callWithLogin(1L, () -> controller.getDeviceInfo("D01"))) + .isInstanceOfSatisfying(ServiceException.class, exception -> { + assertThat(exception.getCode()).isEqualTo(501); + assertThat(exception.getMessage()).isEqualTo("app.device.not.exists.or.denied"); + }); + verifyNoInteractions(deviceCommandService); + } + + @Test + public void getDeviceInfo_sendsQueryPowerCommandForActiveDevice() { + AppController controller = newController(); + AppDeviceVo device = ownedDevice("D01", 1L); + device.setPowerLevel("64"); + when(appDeviceService.queryActiveDevice(1L, "D01")).thenReturn(device); + + R result = callWithLogin(1L, () -> controller.getDeviceInfo("D01")); + + assertThat(result.getCode()).as(result.getMsg()).isEqualTo(200); + assertThat(result.getData()).isSameAs(device); + verify(deviceCommandService).sendQueryPowerCommand("D01", 1L); } @Test diff --git a/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceDataHandlerTest.java b/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceDataHandlerTest.java index a201596..0e85803 100644 --- a/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceDataHandlerTest.java +++ b/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceDataHandlerTest.java @@ -52,7 +52,7 @@ class DeviceDataHandlerTest { when(deviceBindingService.isCurrentBinding("D01", "binding-1")).thenReturn(true); withJsonContext(() -> { - handler.handle("D01", "{\"powerLevel\":\"86\",\"bindingId\":\"binding-1\"}"); + handler.handle("D01", "{\"powerLevel\":\"86\",\"charging\":\"1\",\"bindingId\":\"binding-1\"}"); return null; }); @@ -61,16 +61,62 @@ class DeviceDataHandlerTest { AppDeviceBo updatedDevice = deviceCaptor.getValue(); assertThat(updatedDevice.getDeviceNo()).isEqualTo("D01"); assertThat(updatedDevice.getPowerLevel()).isEqualTo("86"); + assertThat(updatedDevice.getPowerStatus()).isEqualTo("1"); assertThat(updatedDevice.getPowerLevelUpdatatime()).isNotNull(); verify(deviceStatusService).markOnline("D01"); } + @Test + void handleUpdatesPowerReportWithoutBindingIdForActiveBinding() { + when(deviceIdentityResolver.resolveDeviceNo("D01")).thenReturn("D01"); + when(deviceBindingService.isActive("D01")).thenReturn(true); + + withJsonContext(() -> { + handler.handle("D01", "{\"powerLevel\":\"86\"}"); + return null; + }); + + assertPowerUpdate(); + verify(deviceBindingService, never()).isCurrentBinding(anyString(), anyString()); + } + + @Test + void handleUpdatesPowerReportWithBlankBindingIdForActiveBinding() { + when(deviceIdentityResolver.resolveDeviceNo("D01")).thenReturn("D01"); + when(deviceBindingService.isActive("D01")).thenReturn(true); + + withJsonContext(() -> { + handler.handle("D01", "{\"powerLevel\":\"86\",\"bindingId\":\" \"}"); + return null; + }); + + assertPowerUpdate(); + verify(deviceBindingService, never()).isCurrentBinding(anyString(), anyString()); + } + + @Test + void handleUpdatesPowerReportWithNullBindingIdForActiveBinding() { + when(deviceIdentityResolver.resolveDeviceNo("D01")).thenReturn("D01"); + when(deviceBindingService.isActive("D01")).thenReturn(true); + + withJsonContext(() -> { + handler.handle("D01", "{\"powerLevel\":\"86\",\"bindingId\":null}"); + return null; + }); + + assertPowerUpdate(); + verify(deviceBindingService, never()).isCurrentBinding(anyString(), anyString()); + } + @Test void handleIgnoresPowerReportWhenBindingIsNotActive() { when(deviceIdentityResolver.resolveDeviceNo("D01")).thenReturn("D01"); when(deviceBindingService.isActive("D01")).thenReturn(false); - handler.handle("D01", "{\"powerLevel\":\"86\"}"); + withJsonContext(() -> { + handler.handle("D01", "{\"powerLevel\":\"86\"}"); + return null; + }); verifyNoInteractions(appDeviceService, deviceStatusService); } @@ -90,6 +136,16 @@ class DeviceDataHandlerTest { verifyNoInteractions(appDeviceService, deviceStatusService); } + private void assertPowerUpdate() { + ArgumentCaptor deviceCaptor = ArgumentCaptor.forClass(AppDeviceBo.class); + verify(appDeviceService).updateByBo(deviceCaptor.capture()); + AppDeviceBo updatedDevice = deviceCaptor.getValue(); + assertThat(updatedDevice.getDeviceNo()).isEqualTo("D01"); + assertThat(updatedDevice.getPowerLevel()).isEqualTo("86"); + assertThat(updatedDevice.getPowerLevelUpdatatime()).isNotNull(); + verify(deviceStatusService).markOnline("D01"); + } + private T withJsonContext(Supplier action) { try (GenericApplicationContext applicationContext = new GenericApplicationContext()) { applicationContext.registerBean(ObjectMapper.class, (Supplier) ObjectMapper::new); diff --git a/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceStatusHandlerTest.java b/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceStatusHandlerTest.java index 56cccb3..0a21e42 100644 --- a/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceStatusHandlerTest.java +++ b/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceStatusHandlerTest.java @@ -43,6 +43,7 @@ class DeviceStatusHandlerTest { verify(deviceStatusService, never()).markOnline(anyString()); verify(deviceStatusService, never()).markOffline(anyString(), anyString()); + verify(deviceStatusService, never()).markOfflineIfCacheMissing(anyString(), anyString()); } @Test @@ -52,7 +53,8 @@ class DeviceStatusHandlerTest { handler.handle("D01", "{\"status\":\"offline\"}"); - verify(deviceStatusService).markOffline("D01", "设备 MQTT 状态离线"); + verify(deviceStatusService).markOfflineIfCacheMissing("D01", "设备 MQTT 状态离线"); + verify(deviceStatusService, never()).markOffline(anyString(), anyString()); verify(deviceStatusService, never()).markOnline(anyString()); } @@ -65,7 +67,8 @@ class DeviceStatusHandlerTest { handler.handle("unknown-topic-identity", "{\"deviceMac\":\"DC:DA:0C:FA:29:5E\",\"offline\":\"true\"}"); verify(deviceIdentityResolver).resolveDeviceNo("DC:DA:0C:FA:29:5E"); - verify(deviceStatusService).markOffline("D01", "设备 MQTT 状态离线"); + verify(deviceStatusService).markOfflineIfCacheMissing("D01", "设备 MQTT 状态离线"); + verify(deviceStatusService, never()).markOffline(anyString(), anyString()); verify(deviceStatusService, never()).markOnline(anyString()); } @@ -82,7 +85,7 @@ class DeviceStatusHandlerTest { handler.handle(macAddress, "{\"status\":\"offline\"}", false); verify(appDeviceMapper).selectByMac("dc:da:0c:fa:29:5e"); - verify(deviceStatusService).markOffline("D01", "设备 MQTT 状态离线"); + verify(deviceStatusService).markOfflineIfCacheMissing("D01", "设备 MQTT 状态离线"); verify(deviceStatusService, never()).markOffline(macAddress, "设备 MQTT 状态离线"); } @@ -97,6 +100,7 @@ class DeviceStatusHandlerTest { handler.handle(macAddress, "{\"status\":\"offline\"}", false); verify(deviceStatusService, never()).markOffline(anyString(), anyString()); + verify(deviceStatusService, never()).markOfflineIfCacheMissing(anyString(), anyString()); verify(deviceStatusService, never()).markOnline(anyString()); } @@ -109,6 +113,19 @@ class DeviceStatusHandlerTest { verify(deviceStatusService, never()).markOnline(anyString()); verify(deviceStatusService, never()).markOffline(anyString(), anyString()); + verify(deviceStatusService, never()).markOfflineIfCacheMissing(anyString(), anyString()); + } + + @Test + void handleIgnoresRetainedOfflineStatus() { + DeviceStatusHandler handler = newHandler(); + when(deviceIdentityResolver.resolveDeviceNo("D01")).thenReturn("D01"); + + handler.handle("D01", "{\"status\":\"offline\"}", true); + + verify(deviceStatusService, never()).markOffline(anyString(), anyString()); + verify(deviceStatusService, never()).markOfflineIfCacheMissing(anyString(), anyString()); + verify(deviceStatusService, never()).markOnline(anyString()); } private DeviceStatusHandler newHandler() { diff --git a/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceUnbindHandlerTest.java b/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceUnbindHandlerTest.java new file mode 100644 index 0000000..3fcec4d --- /dev/null +++ b/water-modules/water-app/src/test/java/org/dromara/app/handler/DeviceUnbindHandlerTest.java @@ -0,0 +1,87 @@ +package org.dromara.app.handler; + +import cn.hutool.extra.spring.SpringUtil; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.dromara.app.domain.mqtt.DeviceUnbindRequest; +import org.dromara.app.domain.mqtt.DeviceUnbindResult; +import org.dromara.app.service.IAppDeviceBindingService; +import org.dromara.app.service.IDeviceUnbindResultPublisher; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.context.support.GenericApplicationContext; + +import java.util.function.Supplier; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +@Tag("dev") +class DeviceUnbindHandlerTest { + + private static GenericApplicationContext applicationContext; + + @Mock + private DeviceIdentityResolver deviceIdentityResolver; + @Mock + private IAppDeviceBindingService bindingService; + @Mock + private IDeviceUnbindResultPublisher resultPublisher; + + @BeforeAll + static void initializeJsonUtils() { + applicationContext = new GenericApplicationContext(); + applicationContext.registerBean(ObjectMapper.class, (Supplier) ObjectMapper::new); + applicationContext.refresh(); + new SpringUtil().setApplicationContext(applicationContext); + } + + @AfterAll + static void closeApplicationContext() { + applicationContext.close(); + } + + @Test + void handlesPayloadWithoutCommandId() { + DeviceUnbindResult result = new DeviceUnbindResult( + "D01", "binding-1", DeviceUnbindResult.APPLIED, "done"); + when(deviceIdentityResolver.resolveDeviceNo("d01")).thenReturn("D01"); + when(bindingService.unbindByDevice(eq("D01"), any())).thenReturn(result); + + handler().handle("d01", "{\"bindingId\":\"binding-1\",\"reason\":\"long_press\"}"); + + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(DeviceUnbindRequest.class); + verify(bindingService).unbindByDevice(eq("D01"), requestCaptor.capture()); + assertThat(requestCaptor.getValue().getBindingId()).isEqualTo("binding-1"); + assertThat(requestCaptor.getValue().getReason()).isEqualTo("long_press"); + verify(resultPublisher).publish(result); + } + + @Test + void publishesFailureWithoutCommandId() { + when(deviceIdentityResolver.resolveDeviceNo("d01")).thenReturn("D01"); + when(bindingService.unbindByDevice(eq("D01"), any())).thenThrow(new RuntimeException("failed")); + + handler().handle("d01", "{\"bindingId\":\"binding-1\",\"reason\":\"long_press\"}"); + + ArgumentCaptor resultCaptor = ArgumentCaptor.forClass(DeviceUnbindResult.class); + verify(resultPublisher).publish(resultCaptor.capture()); + assertThat(resultCaptor.getValue().deviceNo()).isEqualTo("D01"); + assertThat(resultCaptor.getValue().bindingId()).isEqualTo("binding-1"); + assertThat(resultCaptor.getValue().status()).isEqualTo(DeviceUnbindResult.FAILED); + assertThat(resultCaptor.getValue().message()).isEqualTo("failed"); + } + + private DeviceUnbindHandler handler() { + return new DeviceUnbindHandler(deviceIdentityResolver, bindingService, resultPublisher); + } +} diff --git a/water-modules/water-app/src/test/java/org/dromara/app/handler/MqttDeviceStatusServiceTest.java b/water-modules/water-app/src/test/java/org/dromara/app/handler/MqttDeviceStatusServiceTest.java index d2bac78..492934e 100644 --- a/water-modules/water-app/src/test/java/org/dromara/app/handler/MqttDeviceStatusServiceTest.java +++ b/water-modules/water-app/src/test/java/org/dromara/app/handler/MqttDeviceStatusServiceTest.java @@ -7,8 +7,8 @@ import org.apache.ibatis.builder.MapperBuilderAssistant; import org.apache.ibatis.session.Configuration; import org.dromara.app.domain.AppDevice; import org.dromara.app.mapper.AppDeviceMapper; +import org.dromara.app.service.DeviceOfflineReconcileService; import org.dromara.app.service.IAppDeviceBindingService; -import org.dromara.app.service.IAppWateringLogService; import org.dromara.common.redis.utils.RedisUtils; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; @@ -27,7 +27,9 @@ import java.time.Duration; import java.util.Date; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.*; @@ -40,7 +42,7 @@ class MqttDeviceStatusServiceTest { @Mock private AppDeviceMapper appDeviceMapper; @Mock - private IAppWateringLogService wateringLogService; + private DeviceOfflineReconcileService deviceOfflineReconcileService; @Mock private IAppDeviceBindingService deviceBindingService; @Mock @@ -70,9 +72,22 @@ class MqttDeviceStatusServiceTest { when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); when(lock.isHeldByCurrentThread()).thenReturn(true); + AtomicBoolean databaseUpdated = new AtomicBoolean(); + doAnswer(invocation -> { + databaseUpdated.set(true); + return 1; + }).when(appDeviceMapper).update(eq(null), any(LambdaUpdateWrapper.class)); try (MockedStatic redis = mockStatic(RedisUtils.class)) { redis.when(RedisUtils::getClient).thenReturn(redissonClient); + redis.when(() -> RedisUtils.setCacheObject( + eq("mqtt:device:status:D01"), any(Map.class), eq(Duration.ofSeconds(600)) + )).thenAnswer(invocation -> { + if (!databaseUpdated.get()) { + throw new AssertionError("online cache was written before database status update"); + } + return null; + }); service.markOnline("D01"); @@ -87,21 +102,78 @@ class MqttDeviceStatusServiceTest { } @Test - void markOfflineDeletesOnlineCacheUpdatesDatabaseAndFinishesRunningLogs() throws Exception { + void markOnlineDoesNotWriteCacheWhenDatabaseUpdateFails() throws Exception { MqttDeviceStatusService service = newService(); when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); when(lock.isHeldByCurrentThread()).thenReturn(true); - when(appDeviceMapper.update(eq(null), any(LambdaUpdateWrapper.class))).thenReturn(1); + doThrow(new IllegalStateException("database failure")) + .when(appDeviceMapper).update(eq(null), any(LambdaUpdateWrapper.class)); try (MockedStatic redis = mockStatic(RedisUtils.class)) { redis.when(RedisUtils::getClient).thenReturn(redissonClient); + assertThatThrownBy(() -> service.markOnline("D01")) + .isInstanceOf(IllegalStateException.class) + .hasMessage("database failure"); + + redis.verify(() -> RedisUtils.setCacheObject( + eq("mqtt:device:status:D01"), any(Map.class), eq(Duration.ofSeconds(600)) + ), never()); + verify(lock).unlock(); + } + } + + @Test + void markOfflineDeletesOnlineCacheAfterDatabaseReconciliation() throws Exception { + MqttDeviceStatusService service = newService(); + when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); + when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); + when(lock.isHeldByCurrentThread()).thenReturn(true); + AtomicBoolean reconciliationComplete = new AtomicBoolean(); + doAnswer(invocation -> { + reconciliationComplete.set(true); + return true; + }).when(deviceOfflineReconcileService) + .reconcileOffline(eq("D01"), eq("设备 MQTT 状态离线"), any(Date.class)); + + try (MockedStatic redis = mockStatic(RedisUtils.class)) { + redis.when(RedisUtils::getClient).thenReturn(redissonClient); + redis.when(() -> RedisUtils.deleteObject("mqtt:device:status:D01")) + .thenAnswer(invocation -> { + if (!reconciliationComplete.get()) { + throw new AssertionError("online cache was deleted before database reconciliation"); + } + return null; + }); + service.markOffline("D01", "设备 MQTT 状态离线"); + verify(deviceOfflineReconcileService) + .reconcileOffline(eq("D01"), eq("设备 MQTT 状态离线"), any(Date.class)); redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01")); - verify(appDeviceMapper).update(eq(null), any(LambdaUpdateWrapper.class)); - verify(wateringLogService).finishRunningLogsByDevice(eq("D01"), any(Date.class), eq("设备 MQTT 状态离线")); + verify(lock).unlock(); + } + } + + @Test + void markOfflineKeepsOnlineCacheWhenDatabaseReconciliationFails() throws Exception { + MqttDeviceStatusService service = newService(); + when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); + when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); + when(lock.isHeldByCurrentThread()).thenReturn(true); + doThrow(new IllegalStateException("database failure")) + .when(deviceOfflineReconcileService) + .reconcileOffline(eq("D01"), eq("设备 MQTT 状态离线"), any(Date.class)); + + try (MockedStatic redis = mockStatic(RedisUtils.class)) { + redis.when(RedisUtils::getClient).thenReturn(redissonClient); + + assertThatThrownBy(() -> service.markOffline("D01", "设备 MQTT 状态离线")) + .isInstanceOf(IllegalStateException.class) + .hasMessage("database failure"); + + redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01"), never()); verify(lock).unlock(); } } @@ -122,18 +194,41 @@ class MqttDeviceStatusServiceTest { redis.verify(RedisUtils::getClient); redis.verify(() -> RedisUtils.hasKey("mqtt:device:status:D01")); redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01"), never()); - verify(appDeviceMapper, never()).update(eq(null), any(LambdaUpdateWrapper.class)); - verify(wateringLogService, never()).finishRunningLogsByDevice(eq("D01"), any(Date.class), eq("设备掉线")); + verify(deviceOfflineReconcileService, never()) + .reconcileOffline(eq("D01"), eq("设备掉线"), any(Date.class)); + verify(lock).unlock(); + } + } + + @Test + void markOfflineIfCacheMissingReconcilesInactiveUnboundDevice() throws Exception { + MqttDeviceStatusService service = newService(); + when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); + when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); + when(lock.isHeldByCurrentThread()).thenReturn(true); + when(deviceOfflineReconcileService.reconcileOffline(eq("D01"), eq("设备掉线"), any(Date.class))).thenReturn(true); + + try (MockedStatic redis = mockStatic(RedisUtils.class)) { + redis.when(RedisUtils::getClient).thenReturn(redissonClient); + redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D01")).thenReturn(false); + + boolean reconciled = service.markOfflineIfCacheMissing("D01", "设备掉线"); + + org.assertj.core.api.Assertions.assertThat(reconciled).isTrue(); + verify(deviceOfflineReconcileService) + .reconcileOffline(eq("D01"), eq("设备掉线"), any(Date.class)); + redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01")); + verify(deviceBindingService, never()).isActive("D01"); verify(lock).unlock(); } } private MqttDeviceStatusService newService() { - MqttDeviceStatusService service = new MqttDeviceStatusService(appDeviceMapper, wateringLogService); + MqttDeviceStatusService service = new MqttDeviceStatusService(appDeviceMapper, deviceOfflineReconcileService); ReflectionTestUtils.setField(service, "deviceBindingService", deviceBindingService); ReflectionTestUtils.setField(service, "deviceStatusCachePrefix", "mqtt:device:status:"); ReflectionTestUtils.setField(service, "deviceStatusCacheTtlSeconds", 600L); - when(deviceBindingService.isActive("D01")).thenReturn(true); + lenient().when(deviceBindingService.isActive("D01")).thenReturn(true); return service; } } diff --git a/water-modules/water-app/src/test/java/org/dromara/app/service/DeviceOfflineReconcileServiceTest.java b/water-modules/water-app/src/test/java/org/dromara/app/service/DeviceOfflineReconcileServiceTest.java new file mode 100644 index 0000000..ea5299a --- /dev/null +++ b/water-modules/water-app/src/test/java/org/dromara/app/service/DeviceOfflineReconcileServiceTest.java @@ -0,0 +1,62 @@ +package org.dromara.app.service; + +import org.dromara.app.mapper.AppDeviceMapper; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.transaction.annotation.Transactional; + +import java.lang.reflect.Method; +import java.util.Date; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +@Tag("dev") +class DeviceOfflineReconcileServiceTest { + + @Mock + private AppDeviceMapper appDeviceMapper; + @Mock + private IAppWateringLogService wateringLogService; + + @Test + void reconcileOfflineFinishesLogsWhenDeviceIsAlreadyOffline() { + DeviceOfflineReconcileService service = new DeviceOfflineReconcileService(appDeviceMapper, wateringLogService); + Date offlineTime = new Date(); + when(appDeviceMapper.markOfflineAndStopWork("D01")).thenReturn(0); + when(wateringLogService.finishRunningLogsByDevice("D01", offlineTime, "设备掉线")).thenReturn(2); + + boolean changed = service.reconcileOffline("D01", "设备掉线", offlineTime); + + assertThat(changed).isTrue(); + verify(appDeviceMapper).markOfflineAndStopWork("D01"); + verify(wateringLogService).finishRunningLogsByDevice("D01", offlineTime, "设备掉线"); + } + + @Test + void reconcileOfflineReturnsFalseWhenDatabaseIsAlreadyConsistent() { + DeviceOfflineReconcileService service = new DeviceOfflineReconcileService(appDeviceMapper, wateringLogService); + Date offlineTime = new Date(); + when(appDeviceMapper.markOfflineAndStopWork("D01")).thenReturn(0); + when(wateringLogService.finishRunningLogsByDevice("D01", offlineTime, "设备掉线")).thenReturn(0); + + boolean changed = service.reconcileOffline("D01", "设备掉线", offlineTime); + + assertThat(changed).isFalse(); + } + + @Test + void reconcileOfflineIsPublicAndTransactionalForExceptionRollback() throws Exception { + Method method = DeviceOfflineReconcileService.class + .getMethod("reconcileOffline", String.class, String.class, Date.class); + Transactional transactional = method.getAnnotation(Transactional.class); + + assertThat(transactional).isNotNull(); + assertThat(transactional.rollbackFor()).containsExactly(Exception.class); + } +} \ No newline at end of file diff --git a/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppDeviceBindingServiceImplTest.java b/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppDeviceBindingServiceImplTest.java index a46ad2e..18bdcb4 100644 --- a/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppDeviceBindingServiceImplTest.java +++ b/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppDeviceBindingServiceImplTest.java @@ -134,17 +134,15 @@ class AppDeviceBindingServiceImplTest { @Test void unbindByDeviceCleansCurrentUserDataWithoutSendingInitCommand() { - DeviceUnbindRequest request = request("command-1", "binding-1"); + DeviceUnbindRequest request = request("binding-1"); AppDeviceBinding active = binding(1L, DeviceBindingStatus.ACTIVE, DeviceUnbindSource.DEVICE); AppDeviceBinding unbinding = binding(1L, DeviceBindingStatus.UNBINDING, DeviceUnbindSource.DEVICE); - unbinding.setUnbindCommandId("command-1"); AppDeviceBinding unbound = binding(1L, DeviceBindingStatus.UNBOUND, DeviceUnbindSource.DEVICE); - unbound.setUnbindCommandId("command-1"); AppDevice device = new AppDevice(); device.setDeviceNo("D01"); device.setDeviceName("Garden"); - when(unbindCommandMapper.selectByDeviceAndCommandId("D01", "command-1")).thenReturn(null); + when(unbindCommandMapper.selectByDeviceAndBindingId("D01", "binding-1")).thenReturn(null); when(bindingMapper.selectByDeviceAndBindingId("D01", "binding-1")).thenReturn(active); when(deviceMapper.selectById("D01")).thenReturn(device); when(bindingMapper.update(eq(null), any(Wrapper.class))).thenReturn(1); @@ -153,7 +151,7 @@ class AppDeviceBindingServiceImplTest { DeviceUnbindResult result = service.unbindByDevice("D01", request); assertThat(result.status()).isEqualTo(DeviceUnbindResult.APPLIED); - assertThat(result.commandId()).isEqualTo("command-1"); + assertThat(result.bindingId()).isEqualTo("binding-1"); verify(deviceCommandAckHandler).clearPendingCommands(List.of("D01")); verify(schedulingDeviceMapper).delete(any(Wrapper.class)); verify(wateringLogMapper).delete(any(Wrapper.class)); @@ -164,15 +162,14 @@ class AppDeviceBindingServiceImplTest { } @Test - void unbindByDeviceReturnsAlreadyAppliedForDuplicateCommandId() { + void unbindByDeviceReturnsAlreadyAppliedForDuplicateBindingId() { AppDeviceUnbindCommand command = new AppDeviceUnbindCommand(); command.setDeviceNo("D01"); - command.setCommandId("command-1"); command.setBindingId("binding-1"); command.setCommandStatus(DeviceUnbindCommandStatus.APPLIED); - when(unbindCommandMapper.selectByDeviceAndCommandId("D01", "command-1")).thenReturn(command); + when(unbindCommandMapper.selectByDeviceAndBindingId("D01", "binding-1")).thenReturn(command); - DeviceUnbindResult result = service.unbindByDevice("D01", request("command-1", "binding-1")); + DeviceUnbindResult result = service.unbindByDevice("D01", request("binding-1")); assertThat(result.status()).isEqualTo(DeviceUnbindResult.ALREADY_APPLIED); verifyNoInteractions(bindingMapper, deviceCommandAckHandler); @@ -180,10 +177,10 @@ class AppDeviceBindingServiceImplTest { @Test void unbindByDeviceRejectsStaleBindingId() { - when(unbindCommandMapper.selectByDeviceAndCommandId("D01", "command-1")).thenReturn(null); + when(unbindCommandMapper.selectByDeviceAndBindingId("D01", "old-binding")).thenReturn(null); when(bindingMapper.selectByDeviceAndBindingId("D01", "old-binding")).thenReturn(null); - DeviceUnbindResult result = service.unbindByDevice("D01", request("command-1", "old-binding")); + DeviceUnbindResult result = service.unbindByDevice("D01", request("old-binding")); assertThat(result.status()).isEqualTo(DeviceUnbindResult.STALE_BINDING); ArgumentCaptor commandCaptor = ArgumentCaptor.forClass(AppDeviceUnbindCommand.class); @@ -216,9 +213,8 @@ class AppDeviceBindingServiceImplTest { verify(deviceMapper).update(eq(null), any(Wrapper.class)); } - private DeviceUnbindRequest request(String commandId, String bindingId) { + private DeviceUnbindRequest request(String bindingId) { DeviceUnbindRequest request = new DeviceUnbindRequest(); - request.setCommandId(commandId); request.setBindingId(bindingId); request.setReason("long_press"); return request; diff --git a/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppWateringLogServiceImplTest.java b/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppWateringLogServiceImplTest.java index 958aa81..7995b35 100644 --- a/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppWateringLogServiceImplTest.java +++ b/water-modules/water-app/src/test/java/org/dromara/app/service/impl/AppWateringLogServiceImplTest.java @@ -2,9 +2,9 @@ package org.dromara.app.service.impl; import cn.hutool.extra.spring.SpringUtil; import com.baomidou.mybatisplus.core.conditions.Wrapper; -import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; import io.github.linpeilie.Converter; +import org.apache.ibatis.annotations.Update; import org.apache.ibatis.builder.MapperBuilderAssistant; import org.apache.ibatis.session.Configuration; import org.dromara.app.domain.AppWateringLog; @@ -20,6 +20,7 @@ import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.context.support.GenericApplicationContext; import java.text.SimpleDateFormat; +import java.util.Date; import java.util.List; import java.util.function.Supplier; @@ -312,44 +313,48 @@ class AppWateringLogServiceImplTest { } @Test - void finishRunningLogsByDevice_finishesLatestStatusOneLogWithPlannedEndTime() throws Exception { + void finishRunningLogsByDevice_usesSingleBatchMapperUpdate() throws Exception { SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); - - AppWateringLog runningLog = new AppWateringLog(); - runningLog.setId(105L); - runningLog.setDeviceNo("D01"); - runningLog.setStartTime(format.parse("2026-07-07 09:20:00")); - runningLog.setEndTime(format.parse("2026-07-07 09:30:00")); - runningLog.setDurationMin("10"); - runningLog.setTriggerType("0"); - runningLog.setStatus("1"); - - when(baseMapper.selectOne(any(Wrapper.class))).thenReturn(runningLog); - when(baseMapper.update(eq(null), any(Wrapper.class))).thenReturn(1); + Date endTime = format.parse("2026-07-07 09:23:30"); + when(baseMapper.finishRunningLogsByDevice("D01", endTime, "手动停止浇水")).thenReturn(2); AppWateringLogServiceImpl service = new AppWateringLogServiceImpl(baseMapper); - int result = service.finishRunningLogsByDevice( - "D01", - format.parse("2026-07-07 09:23:30"), - "手动停止浇水" - ); + int result = service.finishRunningLogsByDevice("D01", endTime, "手动停止浇水"); - assertThat(result).isEqualTo(1); - ArgumentCaptor queryCaptor = ArgumentCaptor.forClass(Wrapper.class); - verify(baseMapper).selectOne(queryCaptor.capture()); - String query = queryCaptor.getValue().getCustomSqlSegment().toLowerCase(); - assertThat(query).contains("status ="); - assertThat(query).contains("order by"); - assertThat(query).contains("limit 1"); - assertThat(query).doesNotContain("endtime is null"); + assertThat(result).isEqualTo(2); + verify(baseMapper).finishRunningLogsByDevice("D01", endTime, "手动停止浇水"); + verify(baseMapper, never()).selectOne(any(Wrapper.class)); + verify(baseMapper, never()).update(eq(null), any(Wrapper.class)); + } - ArgumentCaptor updateCaptor = ArgumentCaptor.forClass(Wrapper.class); - verify(baseMapper).update(eq(null), updateCaptor.capture()); - LambdaUpdateWrapper updateWrapper = (LambdaUpdateWrapper) updateCaptor.getValue(); - String sqlSet = updateWrapper.getSqlSet().toLowerCase(); - assertThat(sqlSet).contains("status"); - assertThat(sqlSet).contains("endtime"); - assertThat(updateWrapper.getParamNameValuePairs().values()).contains("0"); + @Test + void finishRunningLogsByDevice_returnsZeroWhenNoRunningLogsRemain() throws Exception { + SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + Date endTime = format.parse("2026-07-07 09:23:30"); + when(baseMapper.finishRunningLogsByDevice("D01", endTime, "重复停止")).thenReturn(0); + + AppWateringLogServiceImpl service = new AppWateringLogServiceImpl(baseMapper); + + assertThat(service.finishRunningLogsByDevice("D01", endTime, "重复停止")).isZero(); + verify(baseMapper).finishRunningLogsByDevice("D01", endTime, "重复停止"); + verify(baseMapper, never()).selectOne(any(Wrapper.class)); + } + + @Test + void finishRunningLogsByDevice_mapperSqlClosesAllRunningLogsWithNonNegativeDuration() throws Exception { + Update update = AppWateringLogMapper.class + .getMethod("finishRunningLogsByDevice", String.class, Date.class, String.class) + .getAnnotation(Update.class); + + assertThat(update).isNotNull(); + String sql = update.value()[0].toLowerCase(); + assertThat(sql).contains("update app_watering_log"); + assertThat(sql).contains("where device_no = #{deviceno}"); + assertThat(sql).contains("and status = '1'"); + assertThat(sql).contains("timestampdiff(minute, start_time, #{endtime})"); + assertThat(sql).contains("greatest("); + assertThat(sql).contains("remark = case when #{remark} is not null"); + assertThat(sql).doesNotContain("limit 1"); } @Test diff --git a/water-modules/water-app/src/test/java/org/dromara/app/task/DeviceOfflineCheckTaskTest.java b/water-modules/water-app/src/test/java/org/dromara/app/task/DeviceOfflineCheckTaskTest.java index 44a1e62..1d66e1a 100644 --- a/water-modules/water-app/src/test/java/org/dromara/app/task/DeviceOfflineCheckTaskTest.java +++ b/water-modules/water-app/src/test/java/org/dromara/app/task/DeviceOfflineCheckTaskTest.java @@ -1,11 +1,7 @@ package org.dromara.app.task; import cn.hutool.extra.spring.SpringUtil; -import com.baomidou.mybatisplus.core.conditions.Wrapper; -import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; -import org.apache.ibatis.builder.MapperBuilderAssistant; -import org.apache.ibatis.session.Configuration; -import org.dromara.app.domain.AppDevice; +import org.apache.ibatis.annotations.Select; import org.dromara.app.handler.MqttDeviceStatusService; import org.dromara.app.mapper.AppDeviceMapper; import org.dromara.common.redis.utils.RedisUtils; @@ -24,7 +20,7 @@ import org.springframework.test.util.ReflectionTestUtils; import java.time.Duration; import java.util.List; -import static org.mockito.ArgumentMatchers.any; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.*; @ExtendWith(MockitoExtension.class) @@ -40,9 +36,6 @@ class DeviceOfflineCheckTaskTest { @BeforeAll static void initializeInfrastructure() { - if (TableInfoHelper.getTableInfo(AppDevice.class) == null) { - TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new Configuration(), ""), AppDevice.class); - } applicationContext = new GenericApplicationContext(); applicationContext.registerBean(RedissonClient.class, () -> mock(RedissonClient.class)); applicationContext.refresh(); @@ -57,9 +50,7 @@ class DeviceOfflineCheckTaskTest { @Test void checkOfflineDevicesAddsTtlToLegacyPermanentOnlineKey() { DeviceOfflineCheckTask task = newTask(); - AppDevice device = new AppDevice(); - device.setDeviceNo("D01"); - when(appDeviceMapper.selectList(any(Wrapper.class))).thenReturn(List.of(device)); + when(appDeviceMapper.selectOfflineReconcileCandidateDeviceNos()).thenReturn(List.of("D01")); try (MockedStatic redis = mockStatic(RedisUtils.class)) { redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D01")).thenReturn(true); @@ -75,9 +66,7 @@ class DeviceOfflineCheckTaskTest { @Test void checkOfflineDevicesMarksDeviceOfflineWhenPowerHeartbeatExpired() { DeviceOfflineCheckTask task = newTask(); - AppDevice device = new AppDevice(); - device.setDeviceNo("D01"); - when(appDeviceMapper.selectList(any(Wrapper.class))).thenReturn(List.of(device)); + when(appDeviceMapper.selectOfflineReconcileCandidateDeviceNos()).thenReturn(List.of("D01")); try (MockedStatic redis = mockStatic(RedisUtils.class)) { redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D01")).thenReturn(false); @@ -88,6 +77,38 @@ class DeviceOfflineCheckTaskTest { } } + @Test + void checkOfflineDevicesContinuesWhenOneDeviceFails() { + DeviceOfflineCheckTask task = newTask(); + when(appDeviceMapper.selectOfflineReconcileCandidateDeviceNos()).thenReturn(List.of("D01", "D02")); + doThrow(new IllegalStateException("first device failed")) + .when(deviceStatusService).markOfflineIfCacheMissing("D01", "设备电量心跳超时"); + + try (MockedStatic redis = mockStatic(RedisUtils.class)) { + redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D01")).thenReturn(false); + redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D02")).thenReturn(false); + + task.checkOfflineDevices(); + + verify(deviceStatusService).markOfflineIfCacheMissing("D01", "设备电量心跳超时"); + verify(deviceStatusService).markOfflineIfCacheMissing("D02", "设备电量心跳超时"); + } + } + + @Test + void candidateQueryCoversDeviceAndRunningLogResidualStates() throws Exception { + Select select = AppDeviceMapper.class + .getMethod("selectOfflineReconcileCandidateDeviceNos") + .getAnnotation(Select.class); + + assertThat(select).isNotNull(); + String sql = select.value()[0].toLowerCase(); + assertThat(sql).contains("coalesce(d.status, '') <> '0'"); + assertThat(sql).contains("coalesce(d.work_status, '') <> '0'"); + assertThat(sql).contains("exists"); + assertThat(sql).contains("l.status = '1'"); + } + private DeviceOfflineCheckTask newTask() { DeviceOfflineCheckTask task = new DeviceOfflineCheckTask(appDeviceMapper, deviceStatusService); ReflectionTestUtils.setField(task, "deviceStatusCachePrefix", "mqtt:device:status:");