设备bug修改 新增设备解绑功能

This commit is contained in:
yuhaiming
2026-09-01 17:24:12 +08:00
parent 509e760f3d
commit 04c83305ad
36 changed files with 796 additions and 214 deletions

Binary file not shown.

View File

@@ -10,7 +10,6 @@ CREATE TABLE app_device_binding (
device_name_snapshot varchar(255) NULL COMMENT '列表显示名称快照', device_name_snapshot varchar(255) NULL COMMENT '列表显示名称快照',
unbound_time datetime NULL COMMENT '解绑完成时间', unbound_time datetime NULL COMMENT '解绑完成时间',
unbind_source varchar(16) NULL COMMENT 'DEVICE/APP/ADMIN', 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_attempts int NOT NULL DEFAULT 0 COMMENT '清理尝试次数',
cleanup_error varchar(500) NULL COMMENT '最近清理错误', cleanup_error varchar(500) NULL COMMENT '最近清理错误',
create_dept bigint NULL, create_dept bigint NULL,
@@ -32,8 +31,7 @@ CREATE TABLE app_device_binding (
CREATE TABLE app_device_unbind_command ( CREATE TABLE app_device_unbind_command (
id bigint NOT NULL, id bigint NOT NULL,
device_no varchar(64) NOT NULL COMMENT '设备编号', 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', command_status varchar(16) NOT NULL COMMENT 'PROCESSING/APPLIED/STALE/FAILED',
message varchar(500) NULL COMMENT '处理结果', message varchar(500) NULL COMMENT '处理结果',
attempt_count int NOT NULL DEFAULT 0 COMMENT '处理次数', attempt_count int NOT NULL DEFAULT 0 COMMENT '处理次数',
@@ -44,9 +42,8 @@ CREATE TABLE app_device_unbind_command (
update_by bigint NULL, update_by bigint NULL,
update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, update_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id), PRIMARY KEY (id),
UNIQUE KEY uk_app_device_unbind_command (device_no, command_id), UNIQUE KEY uk_app_device_unbind_binding (device_no, binding_id)
KEY idx_app_device_unbind_binding (device_no, binding_id) ) COMMENT='设备解绑处理记录';
) COMMENT='设备解绑命令处理记录';
-- Existing bindings must re-establish the device-side bindingId before becoming ACTIVE. -- Existing bindings must re-establish the device-side bindingId before becoming ACTIVE.
INSERT INTO app_device_binding ( INSERT INTO app_device_binding (

View File

@@ -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);

View File

@@ -365,6 +365,11 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
// 其他线程正在刷新,本次跳过;状态写入幂等,下条 ACK 仍会兜底刷新 // 其他线程正在刷新,本次跳过;状态写入幂等,下条 ACK 仍会兜底刷新
return; return;
} }
appDeviceMapper.update(null,
new LambdaUpdateWrapper<AppDevice>()
.set(AppDevice::getStatus, "1")
.eq(AppDevice::getDeviceNo, deviceNo)
);
Map<String, Object> statusCache = new HashMap<>(); Map<String, Object> statusCache = new HashMap<>();
statusCache.put("deviceNo", deviceNo); statusCache.put("deviceNo", deviceNo);
statusCache.put("status", "1"); statusCache.put("status", "1");
@@ -374,11 +379,6 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
statusCache, statusCache,
Duration.ofSeconds(mqttProperties.getCommandAck().getDeviceStatusCacheTtlSeconds()) Duration.ofSeconds(mqttProperties.getCommandAck().getDeviceStatusCacheTtlSeconds())
); );
appDeviceMapper.update(null,
new LambdaUpdateWrapper<AppDevice>()
.set(AppDevice::getStatus, "1")
.eq(AppDevice::getDeviceNo, deviceNo)
);
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
log.warn("[MQTT] 刷新设备在线状态被中断 设备编号={}", deviceNo); log.warn("[MQTT] 刷新设备在线状态被中断 设备编号={}", deviceNo);

View File

@@ -30,7 +30,6 @@ public class MqttDeviceUnbindResultPublisher implements IDeviceUnbindResultPubli
throw new ServiceException("解绑回执缺少设备编号"); throw new ServiceException("解绑回执缺少设备编号");
} }
Map<String, Object> payload = new LinkedHashMap<>(); Map<String, Object> payload = new LinkedHashMap<>();
payload.put("commandId", result.commandId());
payload.put("bindingId", result.bindingId()); payload.put("bindingId", result.bindingId());
payload.put("status", result.status()); payload.put("status", result.status());
payload.put("message", result.message()); payload.put("message", result.message());

View File

@@ -418,10 +418,23 @@ class MqttCommandAckServiceTest {
when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock);
when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true);
when(lock.isHeldByCurrentThread()).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); MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) { try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(RedisUtils::getClient).thenReturn(redissonClient); 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"); ReflectionTestUtils.invokeMethod(service, "refreshDeviceOnline", "D01");
@@ -434,4 +447,30 @@ class MqttCommandAckServiceTest {
verify(lock).unlock(); 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<RedisUtils> 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();
}
}
} }

View File

@@ -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>) 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<String> 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");
}
}

View File

@@ -47,6 +47,12 @@
<groupId>p6spy</groupId> <groupId>p6spy</groupId>
<artifactId>p6spy</artifactId> <artifactId>p6spy</artifactId>
</dependency> </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies> </dependencies>
</project> </project>

View File

@@ -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);
}
}
}

View File

@@ -2,8 +2,8 @@
modulelist=com.baomidou.mybatisplus.extension.p6spy.MybatisPlusLogFactory,com.p6spy.engine.outage.P6OutageFactory modulelist=com.baomidou.mybatisplus.extension.p6spy.MybatisPlusLogFactory,com.p6spy.engine.outage.P6OutageFactory
# 自定义日志打印 # 自定义日志打印
logMessageFormat=com.baomidou.mybatisplus.extension.p6spy.P6SpyLogger logMessageFormat=com.baomidou.mybatisplus.extension.p6spy.P6SpyLogger
#日志输出到控制台 # 日志输出到控制台,定时任务线程不打印 SQL
appender=com.baomidou.mybatisplus.extension.p6spy.StdoutLogger appender=org.dromara.common.mybatis.p6spy.ScheduledTaskAwareStdoutLogger
# 使用日志系统记录 sql # 使用日志系统记录 sql
#appender=com.p6spy.engine.spy.appender.Slf4JLogger #appender=com.p6spy.engine.spy.appender.Slf4JLogger
# 取消JDBC URL前缀 # 取消JDBC URL前缀

View File

@@ -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();
}
}

View File

@@ -130,7 +130,9 @@ public class AppController extends BaseController {
@GetMapping("/device/{deviceNo}") @GetMapping("/device/{deviceNo}")
public R<AppDeviceVo> getDeviceInfo(@NotNull(message = "主键不能为空") public R<AppDeviceVo> getDeviceInfo(@NotNull(message = "主键不能为空")
@PathVariable String deviceNo) { @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(); Long userId = LoginHelper.getUserId();
AppDeviceVo device = appDeviceService.queryActiveDevice(userId, deviceNo); AppDeviceVo device = appDeviceService.queryActiveDevice(userId, deviceNo);
if (device == null) { if (device == null) {
throw new ServiceException("app.device.not.exists.or.denied"); throw new ServiceException("app.device.not.exists.or.denied", 501);
} }
return device; return device;
} }

View File

@@ -28,7 +28,6 @@ public class AppDeviceBinding extends BaseEntity {
private String deviceNameSnapshot; private String deviceNameSnapshot;
private Date unboundTime; private Date unboundTime;
private String unbindSource; private String unbindSource;
private String unbindCommandId;
private Integer cleanupAttempts; private Integer cleanupAttempts;
private String cleanupError; private String cleanupError;
} }

View File

@@ -20,7 +20,6 @@ public class AppDeviceUnbindCommand extends BaseEntity {
@TableId("id") @TableId("id")
private Long id; private Long id;
private String deviceNo; private String deviceNo;
private String commandId;
private String bindingId; private String bindingId;
private String commandStatus; private String commandStatus;
private String message; private String message;

View File

@@ -4,7 +4,6 @@ import lombok.Data;
@Data @Data
public class DeviceUnbindRequest { public class DeviceUnbindRequest {
private String commandId;
private String bindingId; private String bindingId;
private String reason; private String reason;
} }

View File

@@ -2,7 +2,6 @@ package org.dromara.app.domain.mqtt;
public record DeviceUnbindResult( public record DeviceUnbindResult(
String deviceNo, String deviceNo,
String commandId,
String bindingId, String bindingId,
String status, String status,
String message String message

View File

@@ -7,6 +7,7 @@ import org.dromara.app.domain.bo.AppDeviceBo;
import org.dromara.app.mqtt.MqttTopicHandler; import org.dromara.app.mqtt.MqttTopicHandler;
import org.dromara.app.service.IAppDeviceBindingService; import org.dromara.app.service.IAppDeviceBindingService;
import org.dromara.app.service.IAppDeviceService; import org.dromara.app.service.IAppDeviceService;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.json.utils.JsonUtils; import org.dromara.common.json.utils.JsonUtils;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.Lazy;
@@ -49,7 +50,7 @@ public class DeviceDataHandler implements MqttTopicHandler {
return; return;
} }
Map<String, Object> dto = JsonUtils.parseObject(payload, Map.class); Map<String, Object> dto = JsonUtils.parseObject(payload, Map.class);
if (!MqttBindingReportGuard.accepts(deviceBindingService, deviceNo, dto)) { if (!acceptsPowerReport(deviceNo, dto)) {
log.debug("[MQTT] 忽略非当前绑定设备电量上报 设备编号={}", deviceNo); log.debug("[MQTT] 忽略非当前绑定设备电量上报 设备编号={}", deviceNo);
return; return;
} }
@@ -73,4 +74,17 @@ public class DeviceDataHandler implements MqttTopicHandler {
} }
} }
/**
* 电量报文兼容未携带 bindingId 的旧设备,但仍只接受 ACTIVE 绑定设备的上报。
*/
private boolean acceptsPowerReport(String deviceNo, Map<String, Object> 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);
}
} }

View File

@@ -61,8 +61,14 @@ public class DeviceStatusHandler implements MqttTopicHandler {
log.warn("[MQTT] 忽略 retained 离线状态 时间={} 设备编号={} 消息体={}", HandlerLogTime.now(), deviceNo, payload); log.warn("[MQTT] 忽略 retained 离线状态 时间={} 设备编号={} 消息体={}", HandlerLogTime.now(), deviceNo, payload);
return; return;
} }
deviceStatusService.markOffline(deviceNo, OFFLINE_REASON); boolean reconciled = deviceStatusService.markOfflineIfCacheMissing(deviceNo, OFFLINE_REASON);
log.info("[MQTT] 设备状态离线 时间={} 设备编号={} 消息体={}", HandlerLogTime.now(), deviceNo, payload); if (reconciled) {
log.info("[MQTT] 设备在线缓存缺失,已协调离线 时间={} 设备编号={} 消息体={}",
HandlerLogTime.now(), deviceNo, payload);
} else {
log.debug("[MQTT] 设备离线状态未变更,可能仍有在线缓存、已离线或状态锁不可用 时间={} 设备编号={} 消息体={}",
HandlerLogTime.now(), deviceNo, payload);
}
return; return;
} }

View File

@@ -42,19 +42,19 @@ public class DeviceUnbindHandler implements MqttTopicHandler {
} }
DeviceUnbindResult result = bindingService.unbindByDevice(deviceNo, request); DeviceUnbindResult result = bindingService.unbindByDevice(deviceNo, request);
resultPublisher.publish(result); resultPublisher.publish(result);
log.info("[MQTT] 设备解绑已处理 设备编号={} 命令编号={} 结果={}", log.info("[MQTT] 设备解绑已处理 设备编号={} 绑定标识={} 结果={}",
deviceNo, result.commandId(), result.status()); deviceNo, result.bindingId(), result.status());
} catch (RuntimeException e) { } catch (RuntimeException e) {
log.error("[MQTT] 设备解绑处理失败 设备标识={} 设备编号={} 消息体={}", log.error("[MQTT] 设备解绑处理失败 设备标识={} 设备编号={} 消息体={}",
deviceIdentity, deviceNo, payload, e); deviceIdentity, deviceNo, payload, e);
if (deviceNo != null && request != null && request.getCommandId() != null) { if (deviceNo != null && request != null) {
try { try {
resultPublisher.publish(new DeviceUnbindResult( resultPublisher.publish(new DeviceUnbindResult(
deviceNo, request.getCommandId(), request.getBindingId(), deviceNo, request.getBindingId(),
DeviceUnbindResult.FAILED, e.getMessage())); DeviceUnbindResult.FAILED, e.getMessage()));
} catch (RuntimeException publishFailure) { } catch (RuntimeException publishFailure) {
log.warn("[MQTT] 设备解绑失败回执发送失败 设备编号={} 命令编号={}", log.warn("[MQTT] 设备解绑失败回执发送失败 设备编号={} 绑定标识={}",
deviceNo, request.getCommandId(), publishFailure); deviceNo, request.getBindingId(), publishFailure);
} }
} }
} }

View File

@@ -6,8 +6,8 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.dromara.app.domain.AppDevice; import org.dromara.app.domain.AppDevice;
import org.dromara.app.mapper.AppDeviceMapper; import org.dromara.app.mapper.AppDeviceMapper;
import org.dromara.app.service.DeviceOfflineReconcileService;
import org.dromara.app.service.IAppDeviceBindingService; import org.dromara.app.service.IAppDeviceBindingService;
import org.dromara.app.service.IAppWateringLogService;
import org.dromara.common.core.utils.StringUtils; import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.redis.utils.RedisUtils; import org.dromara.common.redis.utils.RedisUtils;
import org.redisson.api.RLock; 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 static final String DEVICE_STATUS_LOCK_PREFIX = "lock:mqtt:device:status:";
private final AppDeviceMapper appDeviceMapper; private final AppDeviceMapper appDeviceMapper;
private final IAppWateringLogService wateringLogService; private final DeviceOfflineReconcileService deviceOfflineReconcileService;
@Lazy @Lazy
@Autowired @Autowired
private IAppDeviceBindingService deviceBindingService; private IAppDeviceBindingService deviceBindingService;
@@ -60,6 +60,11 @@ public class MqttDeviceStatusService {
private void markOnlineInternal(String deviceNo) { private void markOnlineInternal(String deviceNo) {
withDeviceStatusLock(deviceNo, () -> { withDeviceStatusLock(deviceNo, () -> {
appDeviceMapper.update(null,
new LambdaUpdateWrapper<AppDevice>()
.set(AppDevice::getStatus, "1")
.eq(AppDevice::getDeviceNo, deviceNo)
);
Map<String, Object> statusCache = new HashMap<>(); Map<String, Object> statusCache = new HashMap<>();
statusCache.put("deviceNo", deviceNo); statusCache.put("deviceNo", deviceNo);
statusCache.put("status", "1"); statusCache.put("status", "1");
@@ -69,11 +74,6 @@ public class MqttDeviceStatusService {
statusCache, statusCache,
Duration.ofSeconds(deviceStatusCacheTtlSeconds) Duration.ofSeconds(deviceStatusCacheTtlSeconds)
); );
appDeviceMapper.update(null,
new LambdaUpdateWrapper<AppDevice>()
.set(AppDevice::getStatus, "1")
.eq(AppDevice::getDeviceNo, deviceNo)
);
return true; return true;
}); });
} }
@@ -95,28 +95,13 @@ public class MqttDeviceStatusService {
} }
private boolean markOfflineWithLock(String deviceNo, String reason, boolean skipWhenCacheExists) { private boolean markOfflineWithLock(String deviceNo, String reason, boolean skipWhenCacheExists) {
if (!isActive(deviceNo)) {
return false;
}
return withDeviceStatusLock(deviceNo, () -> { return withDeviceStatusLock(deviceNo, () -> {
if (skipWhenCacheExists && RedisUtils.hasKey(deviceStatusCachePrefix + deviceNo)) { if (skipWhenCacheExists && RedisUtils.hasKey(deviceStatusCachePrefix + deviceNo)) {
return false; return false;
} }
boolean reconciled = deviceOfflineReconcileService.reconcileOffline(deviceNo, reason, new Date());
RedisUtils.deleteObject(deviceStatusCachePrefix + deviceNo); RedisUtils.deleteObject(deviceStatusCachePrefix + deviceNo);
int updated = appDeviceMapper.update(null, return reconciled;
new LambdaUpdateWrapper<AppDevice>()
.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;
}); });
} }

View File

@@ -238,6 +238,33 @@ public interface AppDeviceMapper extends BaseMapperPlus<AppDevice, AppDeviceVo>
""") """)
AppDevice selectByMac(@Param("macAddress") String macAddress); 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<String> selectOfflineReconcileCandidateDeviceNos();
@Select(""" @Select("""
select * select *
from app_device from app_device

View File

@@ -10,9 +10,9 @@ public interface AppDeviceUnbindCommandMapper
@Select(""" @Select("""
select * from app_device_unbind_command select * from app_device_unbind_command
where device_no = #{deviceNo} and command_id = #{commandId} where device_no = #{deviceNo} and binding_id = #{bindingId}
limit 1 limit 1
""") """)
AppDeviceUnbindCommand selectByDeviceAndCommandId(@Param("deviceNo") String deviceNo, AppDeviceUnbindCommand selectByDeviceAndBindingId(@Param("deviceNo") String deviceNo,
@Param("commandId") String commandId); @Param("bindingId") String bindingId);
} }

View File

@@ -2,10 +2,13 @@ package org.dromara.app.mapper;
import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select; import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import org.dromara.app.domain.AppWateringLog; import org.dromara.app.domain.AppWateringLog;
import org.dromara.app.domain.vo.AppWateringLogVo; import org.dromara.app.domain.vo.AppWateringLogVo;
import org.dromara.common.mybatis.core.mapper.BaseMapperPlus; import org.dromara.common.mybatis.core.mapper.BaseMapperPlus;
import java.util.Date;
/** /**
* 浇水记录Mapper接口 * 浇水记录Mapper接口
* *
@@ -52,4 +55,17 @@ public interface AppWateringLogMapper extends BaseMapperPlus<AppWateringLog, App
) )
""") """)
Long selectActiveCount(@Param("userId") Long userId); Long selectActiveCount(@Param("userId") Long userId);
@Update("""
update app_watering_log
set end_time = #{endTime},
status = '0',
duration_min = coalesce(greatest(timestampdiff(minute, start_time, #{endTime}), 0), 0),
remark = case when #{remark} is not null and #{remark} <> '' 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);
} }

View File

@@ -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;
}
}

View File

@@ -257,8 +257,8 @@ public class AppDeviceBindingServiceImpl
@Lock4j(keys = {"'device:command:' + #deviceNo"}, acquireTimeout = 5000, expire = 30000) @Lock4j(keys = {"'device:command:' + #deviceNo"}, acquireTimeout = 5000, expire = 30000)
public DeviceUnbindResult unbindByDevice(String deviceNo, DeviceUnbindRequest request) { public DeviceUnbindResult unbindByDevice(String deviceNo, DeviceUnbindRequest request) {
validateDeviceUnbindRequest(deviceNo, request); validateDeviceUnbindRequest(deviceNo, request);
AppDeviceUnbindCommand existing = unbindCommandMapper.selectByDeviceAndCommandId( AppDeviceUnbindCommand existing = unbindCommandMapper.selectByDeviceAndBindingId(
deviceNo, request.getCommandId()); deviceNo, request.getBindingId());
if (existing != null) { if (existing != null) {
if (DeviceUnbindCommandStatus.APPLIED.equals(existing.getCommandStatus())) { if (DeviceUnbindCommandStatus.APPLIED.equals(existing.getCommandStatus())) {
return result(existing, DeviceUnbindResult.ALREADY_APPLIED, "解绑已完成"); return result(existing, DeviceUnbindResult.ALREADY_APPLIED, "解绑已完成");
@@ -272,22 +272,22 @@ public class AppDeviceBindingServiceImpl
? transactionTemplate.execute(status -> acceptDeviceUnbind(deviceNo, request)) ? transactionTemplate.execute(status -> acceptDeviceUnbind(deviceNo, request))
: resumeDeviceUnbind(existing); : resumeDeviceUnbind(existing);
if (acceptance == null) { if (acceptance == null) {
return new DeviceUnbindResult(deviceNo, request.getCommandId(), request.getBindingId(), return new DeviceUnbindResult(deviceNo, request.getBindingId(),
DeviceUnbindResult.FAILED, "解绑处理失败"); DeviceUnbindResult.FAILED, "解绑处理失败");
} }
if (acceptance.stale()) { if (acceptance.stale()) {
return new DeviceUnbindResult(deviceNo, request.getCommandId(), request.getBindingId(), return new DeviceUnbindResult(deviceNo, request.getBindingId(),
DeviceUnbindResult.STALE_BINDING, "绑定标识已过期"); DeviceUnbindResult.STALE_BINDING, "绑定标识已过期");
} }
try { try {
clearRuntimeAfterAccepted(deviceNo); clearRuntimeAfterAccepted(deviceNo);
AppDeviceBinding completed = cleanupBinding(acceptance.binding().getId()); AppDeviceBinding completed = cleanupBinding(acceptance.binding().getId());
return new DeviceUnbindResult(deviceNo, request.getCommandId(), completed.getBindingId(), return new DeviceUnbindResult(deviceNo, completed.getBindingId(),
DeviceUnbindResult.APPLIED, "解绑完成"); DeviceUnbindResult.APPLIED, "解绑完成");
} catch (RuntimeException e) { } catch (RuntimeException e) {
markCleanupFailed(acceptance.binding(), e); markCleanupFailed(acceptance.binding(), e);
return new DeviceUnbindResult(deviceNo, request.getCommandId(), request.getBindingId(), return new DeviceUnbindResult(deviceNo, request.getBindingId(),
DeviceUnbindResult.FAILED, "解绑清理失败,服务器将继续重试"); DeviceUnbindResult.FAILED, "解绑清理失败,服务器将继续重试");
} }
} }
@@ -442,7 +442,6 @@ public class AppDeviceBindingServiceImpl
relationship.setDeviceNameSnapshot(deviceName); relationship.setDeviceNameSnapshot(deviceName);
relationship.setUnboundTime(null); relationship.setUnboundTime(null);
relationship.setUnbindSource(null); relationship.setUnbindSource(null);
relationship.setUnbindCommandId(null);
relationship.setCleanupAttempts(0); relationship.setCleanupAttempts(0);
relationship.setCleanupError(null); relationship.setCleanupError(null);
bindingMapper.updateById(relationship); bindingMapper.updateById(relationship);
@@ -495,7 +494,6 @@ public class AppDeviceBindingServiceImpl
.set(AppDeviceBinding::getDeviceNameSnapshot, .set(AppDeviceBinding::getDeviceNameSnapshot,
device == null ? binding.getDeviceNameSnapshot() : device.getDeviceName()) device == null ? binding.getDeviceNameSnapshot() : device.getDeviceName())
.set(AppDeviceBinding::getUnbindSource, source) .set(AppDeviceBinding::getUnbindSource, source)
.set(AppDeviceBinding::getUnbindCommandId, UUID.randomUUID().toString().replace("-", ""))
.set(AppDeviceBinding::getCleanupAttempts, 0) .set(AppDeviceBinding::getCleanupAttempts, 0)
.set(AppDeviceBinding::getCleanupError, null) .set(AppDeviceBinding::getCleanupError, null)
.eq(AppDeviceBinding::getId, binding.getId()) .eq(AppDeviceBinding::getId, binding.getId())
@@ -523,7 +521,6 @@ public class AppDeviceBindingServiceImpl
.set(AppDeviceBinding::getDeviceNameSnapshot, .set(AppDeviceBinding::getDeviceNameSnapshot,
device == null ? binding.getDeviceNameSnapshot() : device.getDeviceName()) device == null ? binding.getDeviceNameSnapshot() : device.getDeviceName())
.set(AppDeviceBinding::getUnbindSource, DeviceUnbindSource.DEVICE) .set(AppDeviceBinding::getUnbindSource, DeviceUnbindSource.DEVICE)
.set(AppDeviceBinding::getUnbindCommandId, request.getCommandId())
.set(AppDeviceBinding::getCleanupAttempts, 0) .set(AppDeviceBinding::getCleanupAttempts, 0)
.set(AppDeviceBinding::getCleanupError, null) .set(AppDeviceBinding::getCleanupError, null)
.eq(AppDeviceBinding::getId, binding.getId()) .eq(AppDeviceBinding::getId, binding.getId())
@@ -571,7 +568,7 @@ public class AppDeviceBindingServiceImpl
.set(AppDeviceUnbindCommand::getCompletedTime, new Date()) .set(AppDeviceUnbindCommand::getCompletedTime, new Date())
.setSql("attempt_count = attempt_count + 1") .setSql("attempt_count = attempt_count + 1")
.eq(AppDeviceUnbindCommand::getDeviceNo, binding.getDeviceNo()) .eq(AppDeviceUnbindCommand::getDeviceNo, binding.getDeviceNo())
.eq(AppDeviceUnbindCommand::getCommandId, binding.getUnbindCommandId())); .eq(AppDeviceUnbindCommand::getBindingId, binding.getBindingId()));
} }
return bindingMapper.selectById(binding.getId()); return bindingMapper.selectById(binding.getId());
}); });
@@ -609,7 +606,7 @@ public class AppDeviceBindingServiceImpl
.set(AppDeviceUnbindCommand::getMessage, message) .set(AppDeviceUnbindCommand::getMessage, message)
.setSql("attempt_count = attempt_count + 1") .setSql("attempt_count = attempt_count + 1")
.eq(AppDeviceUnbindCommand::getDeviceNo, binding.getDeviceNo()) .eq(AppDeviceUnbindCommand::getDeviceNo, binding.getDeviceNo())
.eq(AppDeviceUnbindCommand::getCommandId, binding.getUnbindCommandId())); .eq(AppDeviceUnbindCommand::getBindingId, binding.getBindingId()));
} }
}); });
AppDeviceBinding current = bindingMapper.selectById(binding.getId()); AppDeviceBinding current = bindingMapper.selectById(binding.getId());
@@ -678,18 +675,17 @@ public class AppDeviceBindingServiceImpl
} }
try { try {
unbindResultPublisher.publish(new DeviceUnbindResult( unbindResultPublisher.publish(new DeviceUnbindResult(
binding.getDeviceNo(), binding.getUnbindCommandId(), binding.getBindingId(), binding.getDeviceNo(), binding.getBindingId(),
DeviceUnbindResult.APPLIED, "解绑完成")); DeviceUnbindResult.APPLIED, "解绑完成"));
} catch (RuntimeException e) { } catch (RuntimeException e) {
log.warn("[设备解绑] 恢复完成但 MQTT 回执发送失败 设备编号={} 命令编号={}", log.warn("[设备解绑] 恢复完成但 MQTT 回执发送失败 设备编号={} 绑定标识={}",
binding.getDeviceNo(), binding.getUnbindCommandId(), e); binding.getDeviceNo(), binding.getBindingId(), e);
} }
} }
private void saveUnbindCommand(String deviceNo, DeviceUnbindRequest request, String status, String message) { private void saveUnbindCommand(String deviceNo, DeviceUnbindRequest request, String status, String message) {
AppDeviceUnbindCommand command = new AppDeviceUnbindCommand(); AppDeviceUnbindCommand command = new AppDeviceUnbindCommand();
command.setDeviceNo(deviceNo); command.setDeviceNo(deviceNo);
command.setCommandId(request.getCommandId());
command.setBindingId(request.getBindingId()); command.setBindingId(request.getBindingId());
command.setCommandStatus(status); command.setCommandStatus(status);
command.setMessage(message); command.setMessage(message);
@@ -700,7 +696,7 @@ public class AppDeviceBindingServiceImpl
try { try {
unbindCommandMapper.insert(command); unbindCommandMapper.insert(command);
} catch (DuplicateKeyException e) { } 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) { private DeviceUnbindResult result(AppDeviceUnbindCommand command, String status, String message) {
return new DeviceUnbindResult(command.getDeviceNo(), command.getCommandId(), return new DeviceUnbindResult(command.getDeviceNo(), command.getBindingId(), status, message);
command.getBindingId(), status, message);
} }
private String abbreviate(String message, int maxLength) { private String abbreviate(String message, int maxLength) {

View File

@@ -285,31 +285,7 @@ public class AppWateringLogServiceImpl implements IAppWateringLogService {
if (StringUtils.isBlank(deviceNo) || endTime == null) { if (StringUtils.isBlank(deviceNo) || endTime == null) {
return 0; return 0;
} }
return baseMapper.finishRunningLogsByDevice(deviceNo, endTime, remark);
AppWateringLog activeLog = baseMapper.selectOne(
Wrappers.<AppWateringLog>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<AppWateringLog> 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);
} }
/** /**

View File

@@ -1,9 +1,7 @@
package org.dromara.app.task; package org.dromara.app.task;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.dromara.app.domain.AppDevice;
import org.dromara.app.handler.MqttDeviceStatusService; import org.dromara.app.handler.MqttDeviceStatusService;
import org.dromara.app.mapper.AppDeviceMapper; import org.dromara.app.mapper.AppDeviceMapper;
import org.dromara.common.redis.utils.RedisUtils; import org.dromara.common.redis.utils.RedisUtils;
@@ -12,7 +10,6 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.time.Duration; import java.time.Duration;
import java.util.ArrayList;
import java.util.List; import java.util.List;
/** /**
@@ -20,8 +17,8 @@ import java.util.List;
* <p> * <p>
* 逻辑:电量上报刷新 Redis 在线 Key,Key 超时后判定设备离线。 * 逻辑:电量上报刷新 Redis 在线 Key,Key 超时后判定设备离线。
* 设备 LWT 离线消息仍可立即更新离线状态,本任务用于处理未收到遗嘱的异常断线。 * 设备 LWT 离线消息仍可立即更新离线状态,本任务用于处理未收到遗嘱的异常断线。
* 本任务定期扫描数据库中非离线状态的设备,如果对应 Redis Key 已不存在, * 本任务定期扫描未离线、未停止工作或存在运行日志的设备,如果对应 Redis Key 已不存在,
* 则将数据库 status 更新为 0(离线)。 * 则协调数据库中的离线状态、工作状态和运行日志。
* </p> * </p>
* *
* <p>扫描间隔通过 {@code mqtt.offline-check.interval-ms} 配置,建议小于设备在线 Key 的 TTL。</p> * <p>扫描间隔通过 {@code mqtt.offline-check.interval-ms} 配置,建议小于设备在线 Key 的 TTL。</p>
@@ -55,33 +52,35 @@ public class DeviceOfflineCheckTask {
return; return;
} }
// 查询数据库中所有非离线状态的设备,避免状态不是 1 但仍需离线修正的设备被漏掉 List<String> candidateDeviceNos = appDeviceMapper.selectOfflineReconcileCandidateDeviceNos();
List<AppDevice> onlineDevices = appDeviceMapper.selectList( if (candidateDeviceNos.isEmpty()) {
new LambdaQueryWrapper<AppDevice>()
.ne(AppDevice::getStatus, "0")
.select(AppDevice::getDeviceNo)
);
if (onlineDevices.isEmpty()) {
return; return;
} }
List<String> offlineDeviceNos = new ArrayList<>(); List<String> updatedDeviceNos = new java.util.ArrayList<>();
int migratedPermanentKeys = 0; int migratedPermanentKeys = 0;
for (AppDevice device : onlineDevices) { for (String deviceNo : candidateDeviceNos) {
String deviceNo = device.getDeviceNo(); if (deviceNo == null || deviceNo.isBlank()) {
String statusKey = deviceStatusCachePrefix + deviceNo;
if (!RedisUtils.hasKey(statusKey)) {
offlineDeviceNos.add(deviceNo);
continue; continue;
} }
try {
long ttl = RedisUtils.getTimeToLive(statusKey); String statusKey = deviceStatusCachePrefix + deviceNo;
if (ttl == -1L) { if (RedisUtils.hasKey(statusKey)) {
RedisUtils.expire(statusKey, Duration.ofSeconds(deviceStatusCacheTtlSeconds)); long ttl = RedisUtils.getTimeToLive(statusKey);
migratedPermanentKeys++; if (ttl == -1L) {
} else if (ttl == -2L) { RedisUtils.expire(statusKey, Duration.ofSeconds(deviceStatusCacheTtlSeconds));
offlineDeviceNos.add(deviceNo); 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); log.info("[设备离线] 已为旧版永久在线缓存补充 TTL,设备数:{}", migratedPermanentKeys);
} }
if (offlineDeviceNos.isEmpty()) {
log.debug("[设备离线] 本次检测未发现离线设备,扫描设备数:{}", onlineDevices.size());
return;
}
List<String> updatedDeviceNos = new ArrayList<>();
for (String deviceNo : offlineDeviceNos) {
if (deviceStatusService.markOfflineIfCacheMissing(deviceNo, "设备电量心跳超时")) {
updatedDeviceNos.add(deviceNo);
}
}
if (!updatedDeviceNos.isEmpty()) { if (!updatedDeviceNos.isEmpty()) {
log.info("[设备离线] 检测到 {} 台设备离线,已更新数据库状态,设备编号:{}", log.info("[设备离线] 检测到 {} 台设备离线,已更新数据库状态,设备编号:{}",
updatedDeviceNos.size(), updatedDeviceNos); updatedDeviceNos.size(), updatedDeviceNos);

View File

@@ -280,6 +280,34 @@ public class AppControllerTest {
assertThatThrownBy(() -> callWithLogin(1L, () -> controller.getDeviceInfo("D01"))) assertThatThrownBy(() -> callWithLogin(1L, () -> controller.getDeviceInfo("D01")))
.isSameAs(failure); .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<AppDeviceVo> 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 @Test

View File

@@ -52,7 +52,7 @@ class DeviceDataHandlerTest {
when(deviceBindingService.isCurrentBinding("D01", "binding-1")).thenReturn(true); when(deviceBindingService.isCurrentBinding("D01", "binding-1")).thenReturn(true);
withJsonContext(() -> { withJsonContext(() -> {
handler.handle("D01", "{\"powerLevel\":\"86\",\"bindingId\":\"binding-1\"}"); handler.handle("D01", "{\"powerLevel\":\"86\",\"charging\":\"1\",\"bindingId\":\"binding-1\"}");
return null; return null;
}); });
@@ -61,16 +61,62 @@ class DeviceDataHandlerTest {
AppDeviceBo updatedDevice = deviceCaptor.getValue(); AppDeviceBo updatedDevice = deviceCaptor.getValue();
assertThat(updatedDevice.getDeviceNo()).isEqualTo("D01"); assertThat(updatedDevice.getDeviceNo()).isEqualTo("D01");
assertThat(updatedDevice.getPowerLevel()).isEqualTo("86"); assertThat(updatedDevice.getPowerLevel()).isEqualTo("86");
assertThat(updatedDevice.getPowerStatus()).isEqualTo("1");
assertThat(updatedDevice.getPowerLevelUpdatatime()).isNotNull(); assertThat(updatedDevice.getPowerLevelUpdatatime()).isNotNull();
verify(deviceStatusService).markOnline("D01"); 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 @Test
void handleIgnoresPowerReportWhenBindingIsNotActive() { void handleIgnoresPowerReportWhenBindingIsNotActive() {
when(deviceIdentityResolver.resolveDeviceNo("D01")).thenReturn("D01"); when(deviceIdentityResolver.resolveDeviceNo("D01")).thenReturn("D01");
when(deviceBindingService.isActive("D01")).thenReturn(false); when(deviceBindingService.isActive("D01")).thenReturn(false);
handler.handle("D01", "{\"powerLevel\":\"86\"}"); withJsonContext(() -> {
handler.handle("D01", "{\"powerLevel\":\"86\"}");
return null;
});
verifyNoInteractions(appDeviceService, deviceStatusService); verifyNoInteractions(appDeviceService, deviceStatusService);
} }
@@ -90,6 +136,16 @@ class DeviceDataHandlerTest {
verifyNoInteractions(appDeviceService, deviceStatusService); verifyNoInteractions(appDeviceService, deviceStatusService);
} }
private void assertPowerUpdate() {
ArgumentCaptor<AppDeviceBo> 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> T withJsonContext(Supplier<T> action) { private <T> T withJsonContext(Supplier<T> action) {
try (GenericApplicationContext applicationContext = new GenericApplicationContext()) { try (GenericApplicationContext applicationContext = new GenericApplicationContext()) {
applicationContext.registerBean(ObjectMapper.class, (Supplier<ObjectMapper>) ObjectMapper::new); applicationContext.registerBean(ObjectMapper.class, (Supplier<ObjectMapper>) ObjectMapper::new);

View File

@@ -43,6 +43,7 @@ class DeviceStatusHandlerTest {
verify(deviceStatusService, never()).markOnline(anyString()); verify(deviceStatusService, never()).markOnline(anyString());
verify(deviceStatusService, never()).markOffline(anyString(), anyString()); verify(deviceStatusService, never()).markOffline(anyString(), anyString());
verify(deviceStatusService, never()).markOfflineIfCacheMissing(anyString(), anyString());
} }
@Test @Test
@@ -52,7 +53,8 @@ class DeviceStatusHandlerTest {
handler.handle("D01", "{\"status\":\"offline\"}"); 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()); 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\"}"); handler.handle("unknown-topic-identity", "{\"deviceMac\":\"DC:DA:0C:FA:29:5E\",\"offline\":\"true\"}");
verify(deviceIdentityResolver).resolveDeviceNo("DC:DA:0C:FA:29:5E"); 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()); verify(deviceStatusService, never()).markOnline(anyString());
} }
@@ -82,7 +85,7 @@ class DeviceStatusHandlerTest {
handler.handle(macAddress, "{\"status\":\"offline\"}", false); handler.handle(macAddress, "{\"status\":\"offline\"}", false);
verify(appDeviceMapper).selectByMac("dc:da:0c:fa:29:5e"); 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 状态离线"); verify(deviceStatusService, never()).markOffline(macAddress, "设备 MQTT 状态离线");
} }
@@ -97,6 +100,7 @@ class DeviceStatusHandlerTest {
handler.handle(macAddress, "{\"status\":\"offline\"}", false); handler.handle(macAddress, "{\"status\":\"offline\"}", false);
verify(deviceStatusService, never()).markOffline(anyString(), anyString()); verify(deviceStatusService, never()).markOffline(anyString(), anyString());
verify(deviceStatusService, never()).markOfflineIfCacheMissing(anyString(), anyString());
verify(deviceStatusService, never()).markOnline(anyString()); verify(deviceStatusService, never()).markOnline(anyString());
} }
@@ -109,6 +113,19 @@ class DeviceStatusHandlerTest {
verify(deviceStatusService, never()).markOnline(anyString()); verify(deviceStatusService, never()).markOnline(anyString());
verify(deviceStatusService, never()).markOffline(anyString(), 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() { private DeviceStatusHandler newHandler() {

View File

@@ -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>) 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<DeviceUnbindRequest> 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<DeviceUnbindResult> 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);
}
}

View File

@@ -7,8 +7,8 @@ import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.ibatis.session.Configuration; import org.apache.ibatis.session.Configuration;
import org.dromara.app.domain.AppDevice; import org.dromara.app.domain.AppDevice;
import org.dromara.app.mapper.AppDeviceMapper; import org.dromara.app.mapper.AppDeviceMapper;
import org.dromara.app.service.DeviceOfflineReconcileService;
import org.dromara.app.service.IAppDeviceBindingService; import org.dromara.app.service.IAppDeviceBindingService;
import org.dromara.app.service.IAppWateringLogService;
import org.dromara.common.redis.utils.RedisUtils; import org.dromara.common.redis.utils.RedisUtils;
import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeAll;
@@ -27,7 +27,9 @@ import java.time.Duration;
import java.util.Date; import java.util.Date;
import java.util.Map; import java.util.Map;
import java.util.concurrent.TimeUnit; 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.any;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*; import static org.mockito.Mockito.*;
@@ -40,7 +42,7 @@ class MqttDeviceStatusServiceTest {
@Mock @Mock
private AppDeviceMapper appDeviceMapper; private AppDeviceMapper appDeviceMapper;
@Mock @Mock
private IAppWateringLogService wateringLogService; private DeviceOfflineReconcileService deviceOfflineReconcileService;
@Mock @Mock
private IAppDeviceBindingService deviceBindingService; private IAppDeviceBindingService deviceBindingService;
@Mock @Mock
@@ -70,9 +72,22 @@ class MqttDeviceStatusServiceTest {
when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock);
when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true);
when(lock.isHeldByCurrentThread()).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<RedisUtils> redis = mockStatic(RedisUtils.class)) { try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(RedisUtils::getClient).thenReturn(redissonClient); 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"); service.markOnline("D01");
@@ -87,21 +102,78 @@ class MqttDeviceStatusServiceTest {
} }
@Test @Test
void markOfflineDeletesOnlineCacheUpdatesDatabaseAndFinishesRunningLogs() throws Exception { void markOnlineDoesNotWriteCacheWhenDatabaseUpdateFails() throws Exception {
MqttDeviceStatusService service = newService(); MqttDeviceStatusService service = newService();
when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock); when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock);
when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true); when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true);
when(lock.isHeldByCurrentThread()).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<RedisUtils> redis = mockStatic(RedisUtils.class)) { try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(RedisUtils::getClient).thenReturn(redissonClient); 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<RedisUtils> 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 状态离线"); service.markOffline("D01", "设备 MQTT 状态离线");
verify(deviceOfflineReconcileService)
.reconcileOffline(eq("D01"), eq("设备 MQTT 状态离线"), any(Date.class));
redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01")); redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01"));
verify(appDeviceMapper).update(eq(null), any(LambdaUpdateWrapper.class)); verify(lock).unlock();
verify(wateringLogService).finishRunningLogsByDevice(eq("D01"), any(Date.class), eq("设备 MQTT 状态离线")); }
}
@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<RedisUtils> 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(); verify(lock).unlock();
} }
} }
@@ -122,18 +194,41 @@ class MqttDeviceStatusServiceTest {
redis.verify(RedisUtils::getClient); redis.verify(RedisUtils::getClient);
redis.verify(() -> RedisUtils.hasKey("mqtt:device:status:D01")); redis.verify(() -> RedisUtils.hasKey("mqtt:device:status:D01"));
redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01"), never()); redis.verify(() -> RedisUtils.deleteObject("mqtt:device:status:D01"), never());
verify(appDeviceMapper, never()).update(eq(null), any(LambdaUpdateWrapper.class)); verify(deviceOfflineReconcileService, never())
verify(wateringLogService, never()).finishRunningLogsByDevice(eq("D01"), any(Date.class), eq("设备掉线")); .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<RedisUtils> 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(); verify(lock).unlock();
} }
} }
private MqttDeviceStatusService newService() { private MqttDeviceStatusService newService() {
MqttDeviceStatusService service = new MqttDeviceStatusService(appDeviceMapper, wateringLogService); MqttDeviceStatusService service = new MqttDeviceStatusService(appDeviceMapper, deviceOfflineReconcileService);
ReflectionTestUtils.setField(service, "deviceBindingService", deviceBindingService); ReflectionTestUtils.setField(service, "deviceBindingService", deviceBindingService);
ReflectionTestUtils.setField(service, "deviceStatusCachePrefix", "mqtt:device:status:"); ReflectionTestUtils.setField(service, "deviceStatusCachePrefix", "mqtt:device:status:");
ReflectionTestUtils.setField(service, "deviceStatusCacheTtlSeconds", 600L); ReflectionTestUtils.setField(service, "deviceStatusCacheTtlSeconds", 600L);
when(deviceBindingService.isActive("D01")).thenReturn(true); lenient().when(deviceBindingService.isActive("D01")).thenReturn(true);
return service; return service;
} }
} }

View File

@@ -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);
}
}

View File

@@ -134,17 +134,15 @@ class AppDeviceBindingServiceImplTest {
@Test @Test
void unbindByDeviceCleansCurrentUserDataWithoutSendingInitCommand() { void unbindByDeviceCleansCurrentUserDataWithoutSendingInitCommand() {
DeviceUnbindRequest request = request("command-1", "binding-1"); DeviceUnbindRequest request = request("binding-1");
AppDeviceBinding active = binding(1L, DeviceBindingStatus.ACTIVE, DeviceUnbindSource.DEVICE); AppDeviceBinding active = binding(1L, DeviceBindingStatus.ACTIVE, DeviceUnbindSource.DEVICE);
AppDeviceBinding unbinding = binding(1L, DeviceBindingStatus.UNBINDING, DeviceUnbindSource.DEVICE); AppDeviceBinding unbinding = binding(1L, DeviceBindingStatus.UNBINDING, DeviceUnbindSource.DEVICE);
unbinding.setUnbindCommandId("command-1");
AppDeviceBinding unbound = binding(1L, DeviceBindingStatus.UNBOUND, DeviceUnbindSource.DEVICE); AppDeviceBinding unbound = binding(1L, DeviceBindingStatus.UNBOUND, DeviceUnbindSource.DEVICE);
unbound.setUnbindCommandId("command-1");
AppDevice device = new AppDevice(); AppDevice device = new AppDevice();
device.setDeviceNo("D01"); device.setDeviceNo("D01");
device.setDeviceName("Garden"); 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(bindingMapper.selectByDeviceAndBindingId("D01", "binding-1")).thenReturn(active);
when(deviceMapper.selectById("D01")).thenReturn(device); when(deviceMapper.selectById("D01")).thenReturn(device);
when(bindingMapper.update(eq(null), any(Wrapper.class))).thenReturn(1); when(bindingMapper.update(eq(null), any(Wrapper.class))).thenReturn(1);
@@ -153,7 +151,7 @@ class AppDeviceBindingServiceImplTest {
DeviceUnbindResult result = service.unbindByDevice("D01", request); DeviceUnbindResult result = service.unbindByDevice("D01", request);
assertThat(result.status()).isEqualTo(DeviceUnbindResult.APPLIED); 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(deviceCommandAckHandler).clearPendingCommands(List.of("D01"));
verify(schedulingDeviceMapper).delete(any(Wrapper.class)); verify(schedulingDeviceMapper).delete(any(Wrapper.class));
verify(wateringLogMapper).delete(any(Wrapper.class)); verify(wateringLogMapper).delete(any(Wrapper.class));
@@ -164,15 +162,14 @@ class AppDeviceBindingServiceImplTest {
} }
@Test @Test
void unbindByDeviceReturnsAlreadyAppliedForDuplicateCommandId() { void unbindByDeviceReturnsAlreadyAppliedForDuplicateBindingId() {
AppDeviceUnbindCommand command = new AppDeviceUnbindCommand(); AppDeviceUnbindCommand command = new AppDeviceUnbindCommand();
command.setDeviceNo("D01"); command.setDeviceNo("D01");
command.setCommandId("command-1");
command.setBindingId("binding-1"); command.setBindingId("binding-1");
command.setCommandStatus(DeviceUnbindCommandStatus.APPLIED); 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); assertThat(result.status()).isEqualTo(DeviceUnbindResult.ALREADY_APPLIED);
verifyNoInteractions(bindingMapper, deviceCommandAckHandler); verifyNoInteractions(bindingMapper, deviceCommandAckHandler);
@@ -180,10 +177,10 @@ class AppDeviceBindingServiceImplTest {
@Test @Test
void unbindByDeviceRejectsStaleBindingId() { 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); 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); assertThat(result.status()).isEqualTo(DeviceUnbindResult.STALE_BINDING);
ArgumentCaptor<AppDeviceUnbindCommand> commandCaptor = ArgumentCaptor.forClass(AppDeviceUnbindCommand.class); ArgumentCaptor<AppDeviceUnbindCommand> commandCaptor = ArgumentCaptor.forClass(AppDeviceUnbindCommand.class);
@@ -216,9 +213,8 @@ class AppDeviceBindingServiceImplTest {
verify(deviceMapper).update(eq(null), any(Wrapper.class)); verify(deviceMapper).update(eq(null), any(Wrapper.class));
} }
private DeviceUnbindRequest request(String commandId, String bindingId) { private DeviceUnbindRequest request(String bindingId) {
DeviceUnbindRequest request = new DeviceUnbindRequest(); DeviceUnbindRequest request = new DeviceUnbindRequest();
request.setCommandId(commandId);
request.setBindingId(bindingId); request.setBindingId(bindingId);
request.setReason("long_press"); request.setReason("long_press");
return request; return request;

View File

@@ -2,9 +2,9 @@ package org.dromara.app.service.impl;
import cn.hutool.extra.spring.SpringUtil; import cn.hutool.extra.spring.SpringUtil;
import com.baomidou.mybatisplus.core.conditions.Wrapper; import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import io.github.linpeilie.Converter; import io.github.linpeilie.Converter;
import org.apache.ibatis.annotations.Update;
import org.apache.ibatis.builder.MapperBuilderAssistant; import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.ibatis.session.Configuration; import org.apache.ibatis.session.Configuration;
import org.dromara.app.domain.AppWateringLog; import org.dromara.app.domain.AppWateringLog;
@@ -20,6 +20,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.context.support.GenericApplicationContext; import org.springframework.context.support.GenericApplicationContext;
import java.text.SimpleDateFormat; import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List; import java.util.List;
import java.util.function.Supplier; import java.util.function.Supplier;
@@ -312,44 +313,48 @@ class AppWateringLogServiceImplTest {
} }
@Test @Test
void finishRunningLogsByDevice_finishesLatestStatusOneLogWithPlannedEndTime() throws Exception { void finishRunningLogsByDevice_usesSingleBatchMapperUpdate() throws Exception {
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date endTime = format.parse("2026-07-07 09:23:30");
AppWateringLog runningLog = new AppWateringLog(); when(baseMapper.finishRunningLogsByDevice("D01", endTime, "手动停止浇水")).thenReturn(2);
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);
AppWateringLogServiceImpl service = new AppWateringLogServiceImpl(baseMapper); AppWateringLogServiceImpl service = new AppWateringLogServiceImpl(baseMapper);
int result = service.finishRunningLogsByDevice( int result = service.finishRunningLogsByDevice("D01", endTime, "手动停止浇水");
"D01",
format.parse("2026-07-07 09:23:30"),
"手动停止浇水"
);
assertThat(result).isEqualTo(1); assertThat(result).isEqualTo(2);
ArgumentCaptor<Wrapper> queryCaptor = ArgumentCaptor.forClass(Wrapper.class); verify(baseMapper).finishRunningLogsByDevice("D01", endTime, "手动停止浇水");
verify(baseMapper).selectOne(queryCaptor.capture()); verify(baseMapper, never()).selectOne(any(Wrapper.class));
String query = queryCaptor.getValue().getCustomSqlSegment().toLowerCase(); verify(baseMapper, never()).update(eq(null), any(Wrapper.class));
assertThat(query).contains("status ="); }
assertThat(query).contains("order by");
assertThat(query).contains("limit 1");
assertThat(query).doesNotContain("endtime is null");
ArgumentCaptor<Wrapper> updateCaptor = ArgumentCaptor.forClass(Wrapper.class); @Test
verify(baseMapper).update(eq(null), updateCaptor.capture()); void finishRunningLogsByDevice_returnsZeroWhenNoRunningLogsRemain() throws Exception {
LambdaUpdateWrapper<AppWateringLog> updateWrapper = (LambdaUpdateWrapper<AppWateringLog>) updateCaptor.getValue(); SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String sqlSet = updateWrapper.getSqlSet().toLowerCase(); Date endTime = format.parse("2026-07-07 09:23:30");
assertThat(sqlSet).contains("status"); when(baseMapper.finishRunningLogsByDevice("D01", endTime, "重复停止")).thenReturn(0);
assertThat(sqlSet).contains("endtime");
assertThat(updateWrapper.getParamNameValuePairs().values()).contains("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 @Test

View File

@@ -1,11 +1,7 @@
package org.dromara.app.task; package org.dromara.app.task;
import cn.hutool.extra.spring.SpringUtil; import cn.hutool.extra.spring.SpringUtil;
import com.baomidou.mybatisplus.core.conditions.Wrapper; import org.apache.ibatis.annotations.Select;
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.dromara.app.handler.MqttDeviceStatusService; import org.dromara.app.handler.MqttDeviceStatusService;
import org.dromara.app.mapper.AppDeviceMapper; import org.dromara.app.mapper.AppDeviceMapper;
import org.dromara.common.redis.utils.RedisUtils; import org.dromara.common.redis.utils.RedisUtils;
@@ -24,7 +20,7 @@ import org.springframework.test.util.ReflectionTestUtils;
import java.time.Duration; import java.time.Duration;
import java.util.List; import java.util.List;
import static org.mockito.ArgumentMatchers.any; import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.*; import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class) @ExtendWith(MockitoExtension.class)
@@ -40,9 +36,6 @@ class DeviceOfflineCheckTaskTest {
@BeforeAll @BeforeAll
static void initializeInfrastructure() { static void initializeInfrastructure() {
if (TableInfoHelper.getTableInfo(AppDevice.class) == null) {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new Configuration(), ""), AppDevice.class);
}
applicationContext = new GenericApplicationContext(); applicationContext = new GenericApplicationContext();
applicationContext.registerBean(RedissonClient.class, () -> mock(RedissonClient.class)); applicationContext.registerBean(RedissonClient.class, () -> mock(RedissonClient.class));
applicationContext.refresh(); applicationContext.refresh();
@@ -57,9 +50,7 @@ class DeviceOfflineCheckTaskTest {
@Test @Test
void checkOfflineDevicesAddsTtlToLegacyPermanentOnlineKey() { void checkOfflineDevicesAddsTtlToLegacyPermanentOnlineKey() {
DeviceOfflineCheckTask task = newTask(); DeviceOfflineCheckTask task = newTask();
AppDevice device = new AppDevice(); when(appDeviceMapper.selectOfflineReconcileCandidateDeviceNos()).thenReturn(List.of("D01"));
device.setDeviceNo("D01");
when(appDeviceMapper.selectList(any(Wrapper.class))).thenReturn(List.of(device));
try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) { try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D01")).thenReturn(true); redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D01")).thenReturn(true);
@@ -75,9 +66,7 @@ class DeviceOfflineCheckTaskTest {
@Test @Test
void checkOfflineDevicesMarksDeviceOfflineWhenPowerHeartbeatExpired() { void checkOfflineDevicesMarksDeviceOfflineWhenPowerHeartbeatExpired() {
DeviceOfflineCheckTask task = newTask(); DeviceOfflineCheckTask task = newTask();
AppDevice device = new AppDevice(); when(appDeviceMapper.selectOfflineReconcileCandidateDeviceNos()).thenReturn(List.of("D01"));
device.setDeviceNo("D01");
when(appDeviceMapper.selectList(any(Wrapper.class))).thenReturn(List.of(device));
try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) { try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
redis.when(() -> RedisUtils.hasKey("mqtt:device:status:D01")).thenReturn(false); 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<RedisUtils> 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() { private DeviceOfflineCheckTask newTask() {
DeviceOfflineCheckTask task = new DeviceOfflineCheckTask(appDeviceMapper, deviceStatusService); DeviceOfflineCheckTask task = new DeviceOfflineCheckTask(appDeviceMapper, deviceStatusService);
ReflectionTestUtils.setField(task, "deviceStatusCachePrefix", "mqtt:device:status:"); ReflectionTestUtils.setField(task, "deviceStatusCachePrefix", "mqtt:device:status:");