opensource/iot-dc3
3
0
Fork 0
mirror of https://gitee.com/pnoker/iot-dc3.git synced 2026-09-01 15:33:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(data): implement RabbitMQ TTL+DLX timeout for driver and device state

Browse Source 
Replace Caffeine LocalCacheService + @Scheduled scanner with
lease-based timeout using RabbitMQ TTL + DLX:
- Driver: each heartbeat publishes a 45s delayed check message;
  DriverTimeoutCheckReceiver performs secondary lease_version and
  expire_time verification before marking offline.
- Device: a self-sustaining 10s tick queue triggers batch scanning
  of expired device leases in EntityStateExpiryScanner.
- Remove OfflineExpiryListener and LocalCacheService from state chain.
- Heartbeat services now only write dc3_entity_state (source of truth).
This commit is contained in:
pnoker
2026-05-22 01:36:14 +08:00
parent 439330f3ff
commit 8772149298
13 changed files with 590 additions and 509 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dc3-common/dc3-common-constant/src/main/java/io/github/pnoker/common/constant/driver/RabbitConstant.java
+26
View File
@@ -93,6 +93,32 @@ public class RabbitConstant {
public static String QUEUE_MQTT = "dc3.q.mqtt";
// State Timeout - Delay Exchange (receives messages to be delayed)
public static String TOPIC_EXCHANGE_STATE_TIMEOUT_DELAY = "dc3.e.state_timeout_delay";
// State Timeout - Check Exchange (receives expired messages from DLX)
public static String TOPIC_EXCHANGE_STATE_TIMEOUT_CHECK = "dc3.e.state_timeout_check";
// Driver timeout delay queue (45s TTL, dead-letter to check exchange)
public static String QUEUE_DRIVER_TIMEOUT_DELAY = "dc3.q.state_timeout.driver.45s";
// Driver timeout check queue (consumed by Data Center)
public static String QUEUE_DRIVER_TIMEOUT_CHECK = "dc3.q.state_timeout.driver_check";
// Routing keys
public static final String ROUTING_DRIVER_TIMEOUT_DELAY = "state.timeout.driver.45s";
public static final String ROUTING_DRIVER_TIMEOUT_CHECK = "state.timeout.driver.check";
// Device scan tick delay queue (10s TTL, dead-letter to scan queue)
public static String QUEUE_DEVICE_SCAN_TICK = "dc3.q.state_timeout.device_scan_tick.10s";
// Device scan execution queue (consumed by Data Center)
public static String QUEUE_DEVICE_SCAN = "dc3.q.state_timeout.device_scan";
// Routing keys
public static final String ROUTING_DEVICE_SCAN_TICK = "state.timeout.device.scan.tick";
public static final String ROUTING_DEVICE_SCAN = "state.timeout.device.scan";
private RabbitConstant() {
throw new IllegalStateException(BaseConstant.UTILITY_CLASS);
}
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/config/DataTopicConfig.java
+31
View File
@@ -164,4 +164,35 @@ public class DataTopicConfig {
return binding;
}
// ===== Device timeout scan tick (TTL + DLX) ===============================
@Bean
Queue deviceScanTickQueue() {
return QueueBuilder.durable(RabbitConstant.QUEUE_DEVICE_SCAN_TICK)
.ttl(10_000) // 10 seconds
.deadLetterExchange(RabbitConstant.TOPIC_EXCHANGE_STATE_TIMEOUT_CHECK)
.deadLetterRoutingKey(RabbitConstant.ROUTING_DEVICE_SCAN)
.build();
}
@Bean
Binding deviceScanTickBinding(Queue deviceScanTickQueue, TopicExchange stateTimeoutDelayExchange) {
return BindingBuilder.bind(deviceScanTickQueue)
.to(stateTimeoutDelayExchange)
.with(RabbitConstant.ROUTING_DEVICE_SCAN_TICK);
}
@Bean
Queue deviceScanQueue() {
return QueueBuilder.durable(RabbitConstant.QUEUE_DEVICE_SCAN).build();
}
@Bean
Binding deviceScanBinding(Queue deviceScanQueue, TopicExchange stateTimeoutCheckExchange) {
return BindingBuilder.bind(deviceScanQueue)
.to(stateTimeoutCheckExchange)
.with(RabbitConstant.ROUTING_DEVICE_SCAN);
}
}
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/biz/impl/DeviceStateServiceImpl.java
+22 -14
View File
@@ -17,16 +17,16 @@
package io.github.pnoker.common.data.biz.impl;
import io.github.pnoker.common.constant.common.PrefixConstant;
import io.github.pnoker.common.data.biz.DeviceAlarmService;
import io.github.pnoker.common.data.biz.DeviceStateService;
import io.github.pnoker.common.data.cache.LocalCacheService;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.entity.dto.DeviceAlarmDTO;
import io.github.pnoker.common.entity.dto.DeviceStateDTO;
import io.github.pnoker.common.entity.ext.JsonExt;
import io.github.pnoker.common.enums.DeviceStatusEnum;
import io.github.pnoker.common.enums.EntityTypeFlagEnum;
import io.github.pnoker.common.enums.TimeoutSourceFlagEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -46,14 +46,16 @@ import java.util.Objects;
@RequiredArgsConstructor
public class DeviceStateServiceImpl implements DeviceStateService {
private final LocalCacheService localCacheService;
private final DeviceAlarmService deviceAlarmService;
private final EntityStateManager entityStateManager;
private static boolean isFlip(String prev, String current) {
return online(prev) != online(current);
private static boolean isFlip(byte prevIndex, String currentCode) {
return online(prevIndex) != online(currentCode);
}
private static boolean online(byte index) {
return index == DeviceStatusEnum.ONLINE.getIndex() || index == DeviceStatusEnum.MAINTAIN.getIndex();
}
private static boolean online(String code) {
@@ -66,8 +68,6 @@ public class DeviceStateServiceImpl implements DeviceStateService {
return;
}
String statusKey = PrefixConstant.DEVICE_STATUS_KEY_PREFIX + entityDTO.getDeviceId();
String prev = localCacheService.getKey(statusKey);
String current = entityDTO.getStatus();
// Persist state lease to database (source of truth)
@@ -81,22 +81,30 @@ public class DeviceStateServiceImpl implements DeviceStateService {
stateDO = new EntityStateDO();
stateDO.setEntityTypeFlag(EntityTypeFlagEnum.DEVICE.getIndex());
stateDO.setEntityId(entityDTO.getDeviceId());
stateDO.setDriverId(Objects.nonNull(entityDTO.getDriverId()) ? entityDTO.getDriverId() : 0L);
stateDO.setParentEntityId(Objects.nonNull(entityDTO.getDriverId()) ? entityDTO.getDriverId() : 0L);
stateDO.setTenantId(entityDTO.getTenantId());
stateDO.setLeaseVersion(1L);
stateDO.setLastStateFlag((byte) DeviceStatusEnum.OFFLINE.getIndex());
stateDO.setLastHeartbeatTime(LocalDateTime.now());
stateDO.setLastAlarmId(0L);
stateDO.setTimeoutSourceFlag((byte) TimeoutSourceFlagEnum.DRIVER.getIndex());
stateDO.setStateExt(JsonExt.builder().type("device-heartbeat").content("").version(1).build());
} else {
stateDO.setLeaseVersion(stateDO.getLeaseVersion() + 1L);
stateDO.setLastStateFlag(stateDO.getStateFlag());
stateDO.setLastHeartbeatTime(LocalDateTime.now());
}
DeviceStatusEnum statusEnum = DeviceStatusEnum.ofCode(current);
stateDO.setStateFlag((byte) (Objects.nonNull(statusEnum) ? statusEnum.getIndex() : 0));
stateDO.setExpireTime(expireTime);
stateDO.setTtlSeconds((int) ttlSeconds);
stateDO.setTimeoutSeconds((int) ttlSeconds);
entityStateManager.saveOrUpdate(stateDO);
localCacheService.setKey(statusKey, current, entityDTO.getTimeOut(), entityDTO.getTimeUnit());
if (Objects.nonNull(prev) && !Objects.equals(prev, current) && isFlip(prev, current)) {
String message = String.format("Device status changed: %s -> %s", prev, current);
byte lastIndex = stateDO.getLastStateFlag();
if (isFlip(lastIndex, current)) {
String message = String.format("Device status changed: %s -> %s",
DeviceStatusEnum.ofIndex(lastIndex) != null ? DeviceStatusEnum.ofIndex(lastIndex).getCode() : "unknown",
current);
DeviceAlarmDTO alarm = DeviceAlarmDTO.builder()
.driverId(entityDTO.getDriverId())
.tenantId(entityDTO.getTenantId())
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/biz/impl/DeviceStatusServiceImpl.java
-2
View File
@@ -18,7 +18,6 @@
package io.github.pnoker.common.data.biz.impl;
import io.github.pnoker.common.data.biz.DeviceStatusService;
import io.github.pnoker.common.data.cache.LocalCacheService;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.data.entity.query.DeviceQuery;
@@ -54,7 +53,6 @@ public class DeviceStatusServiceImpl implements DeviceStatusService {
private final EntityStateManager entityStateManager;
private final LocalCacheService localCacheService;
@Override
public Map<Long, String> getStatusByPage(DeviceQuery pageQuery) {
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/biz/impl/DriverStateServiceImpl.java
+39 -16
View File
@@ -17,23 +17,27 @@
package io.github.pnoker.common.data.biz.impl;
import io.github.pnoker.common.constant.common.PrefixConstant;
import io.github.pnoker.common.constant.driver.RabbitConstant;
import io.github.pnoker.common.data.biz.DriverAlarmService;
import io.github.pnoker.common.data.biz.DriverStateService;
import io.github.pnoker.common.data.cache.LocalCacheService;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.entity.dto.DriverAlarmDTO;
import io.github.pnoker.common.entity.dto.DriverStateDTO;
import io.github.pnoker.common.entity.dto.DriverTimeoutCheckDTO;
import io.github.pnoker.common.entity.ext.JsonExt;
import io.github.pnoker.common.enums.DriverStatusEnum;
import io.github.pnoker.common.enums.EntityTypeFlagEnum;
import io.github.pnoker.common.enums.TimeoutSourceFlagEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
/**
* Business service implementation for driver heartbeat and state processing.
@@ -47,16 +51,20 @@ import java.util.concurrent.TimeUnit;
@RequiredArgsConstructor
public class DriverStateServiceImpl implements DriverStateService {
private static final int STATUS_TTL_SECONDS = 45;
private final LocalCacheService localCacheService;
private static final int STATUS_TIMEOUT_SECONDS = 45;
private final DriverAlarmService driverAlarmService;
private final EntityStateManager entityStateManager;
private static boolean isFlip(String prev, String current) {
return online(prev) != online(current);
private final RabbitTemplate rabbitTemplate;
private static boolean isFlip(byte prevIndex, String currentCode) {
return online(prevIndex) != online(currentCode);
}
private static boolean online(byte index) {
return index == DriverStatusEnum.ONLINE.getIndex() || index == DriverStatusEnum.MAINTAIN.getIndex();
}
private static boolean online(String code) {
@@ -69,8 +77,6 @@ public class DriverStateServiceImpl implements DriverStateService {
return;
}
String statusKey = PrefixConstant.DRIVER_STATUS_KEY_PREFIX + entityDTO.getDriverId();
String prev = localCacheService.getKey(statusKey);
String current = entityDTO.getStatus();
// Persist state lease to database (source of truth)
@@ -78,27 +84,44 @@ public class DriverStateServiceImpl implements DriverStateService {
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DRIVER.getIndex())
.eq(EntityStateDO::getEntityId, entityDTO.getDriverId())
.one();
LocalDateTime expireTime = LocalDateTime.now().plusSeconds(STATUS_TTL_SECONDS);
LocalDateTime expireTime = LocalDateTime.now().plusSeconds(STATUS_TIMEOUT_SECONDS);
if (Objects.isNull(stateDO)) {
stateDO = new EntityStateDO();
stateDO.setEntityTypeFlag(EntityTypeFlagEnum.DRIVER.getIndex());
stateDO.setEntityId(entityDTO.getDriverId());
stateDO.setDriverId(entityDTO.getDriverId());
stateDO.setParentEntityId(entityDTO.getDriverId());
stateDO.setTenantId(entityDTO.getTenantId());
stateDO.setLeaseVersion(1L);
stateDO.setLastStateFlag((byte) DriverStatusEnum.OFFLINE.getIndex());
stateDO.setLastHeartbeatTime(LocalDateTime.now());
stateDO.setLastAlarmId(0L);
stateDO.setTimeoutSourceFlag((byte) TimeoutSourceFlagEnum.SYSTEM.getIndex());
stateDO.setStateExt(JsonExt.builder().type("driver-heartbeat").content("").version(1).build());
} else {
stateDO.setLeaseVersion(stateDO.getLeaseVersion() + 1L);
stateDO.setLastStateFlag(stateDO.getStateFlag());
stateDO.setLastHeartbeatTime(LocalDateTime.now());
}
DriverStatusEnum statusEnum = DriverStatusEnum.ofCode(current);
stateDO.setStateFlag((byte) (Objects.nonNull(statusEnum) ? statusEnum.getIndex() : 0));
stateDO.setExpireTime(expireTime);
stateDO.setTtlSeconds(STATUS_TTL_SECONDS);
stateDO.setTimeoutSeconds(STATUS_TIMEOUT_SECONDS);
entityStateManager.saveOrUpdate(stateDO);
localCacheService.setKey(statusKey, current, STATUS_TTL_SECONDS, TimeUnit.SECONDS);
// Publish timeout check message with current lease version
DriverTimeoutCheckDTO checkDTO = DriverTimeoutCheckDTO.builder()
.driverId(entityDTO.getDriverId())
.leaseVersion(stateDO.getLeaseVersion())
.tenantId(entityDTO.getTenantId())
.build();
rabbitTemplate.convertAndSend(RabbitConstant.TOPIC_EXCHANGE_STATE_TIMEOUT_DELAY,
RabbitConstant.ROUTING_DRIVER_TIMEOUT_DELAY, checkDTO);
if (Objects.nonNull(prev) && !Objects.equals(prev, current) && isFlip(prev, current)) {
String message = String.format("Driver status changed: %s -> %s", prev, current);
byte lastIndex = stateDO.getLastStateFlag();
if (isFlip(lastIndex, current)) {
String message = String.format("Driver status changed: %s -> %s",
DriverStatusEnum.ofIndex(lastIndex) != null ? DriverStatusEnum.ofIndex(lastIndex).getCode() : "unknown",
current);
DriverAlarmDTO alarm = DriverAlarmDTO.builder()
.tenantId(entityDTO.getTenantId())
.driverId(entityDTO.getDriverId())
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/biz/impl/DriverStatusServiceImpl.java
-2
View File
@@ -18,7 +18,6 @@
package io.github.pnoker.common.data.biz.impl;
import io.github.pnoker.common.data.biz.DriverStatusService;
import io.github.pnoker.common.data.cache.LocalCacheService;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.data.entity.query.DriverQuery;
@@ -60,7 +59,6 @@ public class DriverStatusServiceImpl implements DriverStatusService {
private final EntityStateManager entityStateManager;
private final LocalCacheService localCacheService;
@Override
public Map<Long, String> getStatusByPage(DriverQuery pageQuery) {
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/biz/impl/EntityStateExpiryScanner.java
+94 -87
View File
@@ -17,24 +17,29 @@
package io.github.pnoker.common.data.biz.impl;
import io.github.pnoker.common.constant.driver.RabbitConstant;
import io.github.pnoker.common.data.biz.alarm.AlarmRuleTriggerService;
import io.github.pnoker.common.data.dal.EntityAlarmManager;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityAlarmDO;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.entity.dto.DeviceAlarmDTO;
import io.github.pnoker.common.entity.dto.DriverAlarmDTO;
import io.github.pnoker.common.entity.ext.JsonExt;
import io.github.pnoker.common.enums.AlarmMessageLevelFlagEnum;
import io.github.pnoker.common.enums.AlarmSourceFlagEnum;
import io.github.pnoker.common.enums.AlarmTargetTypeFlagEnum;
import io.github.pnoker.common.enums.AlarmTypeFlagEnum;
import io.github.pnoker.common.enums.DeviceStatusEnum;
import io.github.pnoker.common.enums.DriverStatusEnum;
import io.github.pnoker.common.enums.EntityTypeFlagEnum;
import io.github.pnoker.common.utils.RabbitAckUtil;
import com.rabbitmq.client.Channel;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
@@ -42,16 +47,16 @@ import java.util.List;
import java.util.Objects;
/**
* Scans {@code dc3_entity_state} for expired leases and generates offline alarm
* records. This replaces the Caffeine-based {@code OfflineExpiryListener} as the
* primary expiry detection mechanism, surviving restarts and working consistently
* across multiple Data Center instances.
* Tick-triggered scanner for expired device state leases.
*
* <p>The scanner runs every 15 seconds. For each expired row it uses an atomic
* {@code lambdaUpdate} with a {@code lease_version} WHERE condition so that only
* one Data Center instance processes a given expiry — if another instance (or a
* late heartbeat) already updated the row, the UPDATE affects zero rows and the
* alarm is skipped.
* <p>A RabbitMQ TTL + DLX tick queue fires every 10 seconds. On each tick the
* scanner queries {@code dc3_entity_state} for devices whose
* {@code expire_time <= now()} and whose {@code state_flag} is still in the
* online family. Each expired device is atomically claimed via
* {@code lease_version} and an offline alarm is written.
*
* <p>After processing, the scanner publishes the next tick so the cycle
* continues indefinitely.
*
* @author pnoker
* @version 2026.5.22
@@ -64,14 +69,56 @@ public class EntityStateExpiryScanner {
private static final int OFFLINE_RENEW_SECONDS = 300;
private static final int BATCH_LIMIT = 500;
private final EntityStateManager entityStateManager;
private final EntityAlarmManager entityAlarmManager;
private final AlarmRuleTriggerService alarmRuleTriggerService;
private final RabbitTemplate rabbitTemplate;
@Scheduled(fixedDelay = 15_000, initialDelay = 30_000)
public void scanExpiredLeases() {
/**
* Bootstrap the first tick on startup. Subsequent ticks are self-sustaining:
* each scan cycle publishes the next tick.
*/
@PostConstruct
void publishInitialTick() {
rabbitTemplate.convertAndSend(
RabbitConstant.TOPIC_EXCHANGE_STATE_TIMEOUT_DELAY,
RabbitConstant.ROUTING_DEVICE_SCAN_TICK,
"tick");
log.info("Published initial device scan tick");
}
/**
* Process one scan cycle: find expired devices, mark offline, write alarms,
* then publish the next tick.
*/
@RabbitHandler
@RabbitListener(queues = "#{deviceScanQueue.name}")
public void onScanTick(Channel channel, Message message) {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
scanExpiredDevices();
// Publish next tick to keep the cycle going
rabbitTemplate.convertAndSend(
RabbitConstant.TOPIC_EXCHANGE_STATE_TIMEOUT_DELAY,
RabbitConstant.ROUTING_DEVICE_SCAN_TICK,
"tick");
RabbitAckUtil.ack(channel, deliveryTag);
} catch (Exception e) {
log.error("Device scan tick failed", e);
RabbitAckUtil.nack(channel, deliveryTag, true);
}
}
private void scanExpiredDevices() {
LocalDateTime now = LocalDateTime.now();
List<EntityStateDO> expired = entityStateManager.lambdaQuery()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DEVICE.getIndex())
.in(EntityStateDO::getStateFlag,
DeviceStatusEnum.ONLINE.getIndex(),
DeviceStatusEnum.MAINTAIN.getIndex())
.lt(EntityStateDO::getExpireTime, now)
.last("LIMIT " + BATCH_LIMIT)
.list();
@@ -82,31 +129,19 @@ public class EntityStateExpiryScanner {
for (EntityStateDO state : expired) {
try {
processExpiredState(state);
processExpiredDevice(state);
} catch (Exception e) {
log.warn("Expiry processing failed, entity_type={}, entity_id={}",
state.getEntityTypeFlag(), state.getEntityId(), e);
log.warn("Device expiry processing failed, deviceId={}", state.getEntityId(), e);
}
}
}
private void processExpiredState(EntityStateDO scanned) {
EntityTypeFlagEnum typeFlag = EntityTypeFlagEnum.ofIndex(scanned.getEntityTypeFlag());
if (Objects.isNull(typeFlag)) {
return;
}
byte offlineIndex;
if (typeFlag == EntityTypeFlagEnum.DRIVER) {
offlineIndex = (byte) DriverStatusEnum.OFFLINE.getIndex();
} else {
offlineIndex = (byte) DeviceStatusEnum.OFFLINE.getIndex();
}
private void processExpiredDevice(EntityStateDO scanned) {
byte offlineIndex = (byte) DeviceStatusEnum.OFFLINE.getIndex();
// Already offline — just push expire_time forward
if (Objects.equals(scanned.getStateFlag(), offlineIndex)) {
entityStateManager.lambdaUpdate()
.eq(EntityStateDO::getEntityTypeFlag, scanned.getEntityTypeFlag())
.eq(EntityStateDO::getEntityId, scanned.getEntityId())
.eq(EntityStateDO::getLeaseVersion, scanned.getLeaseVersion())
.set(EntityStateDO::getLeaseVersion, scanned.getLeaseVersion() + 1L)
@@ -115,15 +150,16 @@ public class EntityStateExpiryScanner {
return;
}
// Atomically claim: UPDATE ... WHERE lease_version = scanned version
// Atomically claim
long newVersion = scanned.getLeaseVersion() + 1L;
LocalDateTime renewTime = LocalDateTime.now().plusSeconds(OFFLINE_RENEW_SECONDS);
boolean claimed = entityStateManager.lambdaUpdate()
.eq(EntityStateDO::getEntityTypeFlag, scanned.getEntityTypeFlag())
.eq(EntityStateDO::getEntityId, scanned.getEntityId())
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DEVICE.getIndex())
.eq(EntityStateDO::getLeaseVersion, scanned.getLeaseVersion())
.set(EntityStateDO::getLeaseVersion, newVersion)
.set(EntityStateDO::getStateFlag, offlineIndex)
.set(EntityStateDO::getLastStateFlag, scanned.getStateFlag())
.set(EntityStateDO::getExpireTime, renewTime)
.update();
if (!claimed) {
@@ -131,76 +167,47 @@ public class EntityStateExpiryScanner {
}
// Write alarm row
String prevStatusName;
if (typeFlag == EntityTypeFlagEnum.DRIVER) {
DriverStatusEnum prev = DriverStatusEnum.ofIndex(scanned.getStateFlag());
prevStatusName = Objects.nonNull(prev) ? prev.getCode() : "unknown";
} else {
DeviceStatusEnum prev = DeviceStatusEnum.ofIndex(scanned.getStateFlag());
prevStatusName = Objects.nonNull(prev) ? prev.getCode() : "unknown";
}
DeviceStatusEnum prev = DeviceStatusEnum.ofIndex(scanned.getStateFlag());
String prevCode = Objects.nonNull(prev) ? prev.getCode() : "unknown";
String message = String.format("Device heartbeat timed out (last=%s); marked OFFLINE", prevCode);
String message = String.format("%s heartbeat timed out (last=%s); marked OFFLINE",
typeFlag.getCode(), prevStatusName);
EntityAlarmDO alarm = new EntityAlarmDO();
alarm.setAlarmTargetTypeFlag(AlarmTargetTypeFlagEnum.DEVICE.getIndex());
alarm.setEntityId(scanned.getEntityId());
alarm.setDriverId(scanned.getDriverId());
if (typeFlag == EntityTypeFlagEnum.DRIVER) {
alarm.setAlarmTargetTypeFlag(AlarmTargetTypeFlagEnum.DRIVER.getIndex());
alarm.setDeviceId(0L);
alarm.setAlarmExt(JsonExt.builder()
.type("driver-offline")
.content(message)
.version(1)
.build());
} else {
alarm.setAlarmTargetTypeFlag(AlarmTargetTypeFlagEnum.DEVICE.getIndex());
alarm.setDeviceId(scanned.getEntityId());
alarm.setAlarmExt(JsonExt.builder()
.type("device-offline")
.content(message)
.version(1)
.build());
}
alarm.setDriverId(scanned.getParentEntityId());
alarm.setDeviceId(scanned.getEntityId());
alarm.setPointId(0L);
alarm.setRuleId(0L);
alarm.setRuleStateId(0L);
alarm.setAlarmTypeFlag(AlarmTypeFlagEnum.OFFLINE.getIndex());
alarm.setAlarmSourceFlag(AlarmSourceFlagEnum.STATE_TIMEOUT.getIndex());
alarm.setAlarmLevelFlag(AlarmMessageLevelFlagEnum.P1.getIndex());
alarm.setAlarmExt(JsonExt.builder().type("device-offline").content(message).version(1).build());
alarm.setExpiredTime(0L);
alarm.setConfirmFlag((byte) 0);
alarm.setTenantId(scanned.getTenantId());
entityAlarmManager.save(alarm);
// Update lastAlarmId
entityStateManager.lambdaUpdate()
.eq(EntityStateDO::getEntityId, scanned.getEntityId())
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DEVICE.getIndex())
.set(EntityStateDO::getLastAlarmId, alarm.getId())
.update();
// Trigger alarm rule pipeline
if (typeFlag == EntityTypeFlagEnum.DRIVER) {
DriverAlarmDTO dto = DriverAlarmDTO.builder()
.tenantId(scanned.getTenantId())
.driverId(scanned.getEntityId())
.status(null)
.statusName(null)
.message(message)
.alarmId(alarm.getId())
.build();
alarmRuleTriggerService.processDriverAlarm(dto);
} else {
DeviceAlarmDTO dto = DeviceAlarmDTO.builder()
.driverId(scanned.getDriverId())
.tenantId(scanned.getTenantId())
.deviceId(scanned.getEntityId())
.status(null)
.statusName(null)
.message(message)
.alarmId(alarm.getId())
.build();
alarmRuleTriggerService.processDeviceAlarm(dto);
}
DeviceAlarmDTO dto = DeviceAlarmDTO.builder()
.driverId(scanned.getParentEntityId())
.tenantId(scanned.getTenantId())
.deviceId(scanned.getEntityId())
.status(DeviceStatusEnum.OFFLINE.getCode())
.statusName(DeviceStatusEnum.OFFLINE.name())
.message(message)
.alarmId(alarm.getId())
.build();
alarmRuleTriggerService.processDeviceAlarm(dto);
log.info("State lease expired: type={}, entityId={}, tenantId={}, prevStatus={}",
typeFlag.getCode(), scanned.getEntityId(), scanned.getTenantId(), prevStatusName);
log.info("Device scan marked OFFLINE: deviceId={}, tenantId={}, prevStatus={}",
scanned.getEntityId(), scanned.getTenantId(), prevCode);
}
}
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/biz/impl/OfflineExpiryListener.java
-252
View File
@@ -1,252 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.data.biz.impl;
import io.github.pnoker.common.constant.common.PrefixConstant;
import io.github.pnoker.common.data.biz.alarm.AlarmRuleTriggerService;
import io.github.pnoker.common.data.cache.LocalCacheService;
import io.github.pnoker.common.data.dal.EntityAlarmManager;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityAlarmDO;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.entity.dto.DeviceAlarmDTO;
import io.github.pnoker.common.entity.dto.DriverAlarmDTO;
import io.github.pnoker.common.entity.ext.JsonExt;
import io.github.pnoker.common.enums.AlarmMessageLevelFlagEnum;
import io.github.pnoker.common.enums.AlarmSourceFlagEnum;
import io.github.pnoker.common.enums.AlarmTargetTypeFlagEnum;
import io.github.pnoker.common.enums.AlarmTypeFlagEnum;
import io.github.pnoker.common.enums.DeviceStatusEnum;
import io.github.pnoker.common.enums.DriverStatusEnum;
import io.github.pnoker.common.enums.EntityTypeFlagEnum;
import io.github.pnoker.common.facade.api.DeviceFacade;
import io.github.pnoker.common.facade.api.DriverFacade;
import io.github.pnoker.common.facade.entity.bo.FacadeDeviceBO;
import io.github.pnoker.common.facade.entity.bo.FacadeDriverBO;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
/**
* Turns expired online-status cache keys into OFFLINE alarm rows in
* {@code dc3_entity_alarm}. Without this a driver or device that stops sending
* heartbeats would silently drop off the dashboard.
*
* @author pnoker
* @version 2025.9.0
* @since 2026.5.2
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class OfflineExpiryListener {
private final LocalCacheService localCacheService;
private final DriverFacade driverFacade;
private final DeviceFacade deviceFacade;
private final EntityAlarmManager entityAlarmManager;
private final AlarmRuleTriggerService alarmRuleTriggerService;
private final EntityStateManager entityStateManager;
private static Long parseIdSuffix(String key, String prefix) {
try {
return Long.parseLong(key.substring(prefix.length()));
} catch (Exception e) {
log.debug("Unexpected status key '{}': {}", key, e.getMessage());
return null;
}
}
@PostConstruct
void register() {
localCacheService.onExpire(this::onExpire);
}
private void onExpire(String key, Object lastValue) {
if (Objects.isNull(key))
return;
final String lastStatus = lastValue instanceof String s ? s : null;
CompletableFuture.runAsync(() -> {
try {
if (key.startsWith(PrefixConstant.DRIVER_STATUS_KEY_PREFIX)) {
handleDriverExpiry(key, lastStatus);
} else if (key.startsWith(PrefixConstant.DEVICE_STATUS_KEY_PREFIX)) {
handleDeviceExpiry(key, lastStatus);
}
} catch (Exception e) {
log.warn("Offline expiry handling failed, key={}", key, e);
}
});
}
private void handleDriverExpiry(String key, String lastStatus) {
if (Objects.equals(lastStatus, DriverStatusEnum.OFFLINE.getCode()))
return;
Long id = parseIdSuffix(key, PrefixConstant.DRIVER_STATUS_KEY_PREFIX);
// Skip if the DB scanner has already processed this expiry
EntityStateDO dbState = entityStateManager.lambdaQuery()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DRIVER.getIndex())
.eq(EntityStateDO::getEntityId, id)
.one();
if (Objects.nonNull(dbState) && Objects.equals(dbState.getStateFlag(), DriverStatusEnum.OFFLINE.getIndex())) {
return;
}
if (Objects.isNull(id))
return;
// Mark DB state as offline so the scanner does not duplicate the alarm
if (Objects.nonNull(dbState)) {
entityStateManager.lambdaUpdate()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DRIVER.getIndex())
.eq(EntityStateDO::getEntityId, id)
.eq(EntityStateDO::getLeaseVersion, dbState.getLeaseVersion())
.set(EntityStateDO::getStateFlag, (byte) DriverStatusEnum.OFFLINE.getIndex())
.set(EntityStateDO::getLeaseVersion, dbState.getLeaseVersion() + 1L)
.set(EntityStateDO::getExpireTime, LocalDateTime.now().plusSeconds(300))
.update();
}
FacadeDriverBO driver = driverFacade.getById(id);
if (Objects.isNull(driver)) {
log.debug("Driver {} not found when handling offline expiry", id);
return;
}
if (Objects.isNull(driver.getTenantId()) || driver.getTenantId() <= 0) {
log.warn("Drop driver offline alarm because tenantId could not be resolved, driverId={}", id);
return;
}
String message = String.format("Driver heartbeat timed out (last=%s); marked OFFLINE", lastStatus);
EntityAlarmDO entity = new EntityAlarmDO();
entity.setAlarmTargetTypeFlag(AlarmTargetTypeFlagEnum.DRIVER.getIndex());
entity.setEntityId(id);
entity.setDriverId(id);
entity.setDeviceId(0L);
entity.setPointId(0L);
entity.setRuleId(0L);
entity.setAlarmTypeFlag(AlarmTypeFlagEnum.OFFLINE.getIndex());
entity.setAlarmSourceFlag(AlarmSourceFlagEnum.STATE_TIMEOUT.getIndex());
// Heartbeat-timeout offline events default to P1 — they indicate a
// connectivity problem the operator usually wants to see ahead of normal
// P2 device-reported issues.
entity.setAlarmLevelFlag(AlarmMessageLevelFlagEnum.P1.getIndex());
entity.setAlarmExt(JsonExt.builder()
.type("driver-offline")
.content(message)
.version(1)
.build());
entity.setExpiredTime(0L);
entity.setConfirmFlag((byte) 0);
entity.setTenantId(driver.getTenantId());
entityAlarmManager.save(entity);
DriverAlarmDTO alarm = DriverAlarmDTO.builder()
.tenantId(driver.getTenantId())
.driverId(id)
.status(null)
.statusName(null)
.message(message)
.alarmId(entity.getId())
.build();
alarmRuleTriggerService.processDriverAlarm(alarm);
}
private void handleDeviceExpiry(String key, String lastStatus) {
if (Objects.equals(lastStatus, DeviceStatusEnum.OFFLINE.getCode()))
return;
Long id = parseIdSuffix(key, PrefixConstant.DEVICE_STATUS_KEY_PREFIX);
// Skip if the DB scanner has already processed this expiry
EntityStateDO dbState = entityStateManager.lambdaQuery()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DEVICE.getIndex())
.eq(EntityStateDO::getEntityId, id)
.one();
if (Objects.nonNull(dbState) && Objects.equals(dbState.getStateFlag(), DeviceStatusEnum.OFFLINE.getIndex())) {
return;
}
if (Objects.isNull(id))
return;
// Mark DB state as offline so the scanner does not duplicate the alarm
if (Objects.nonNull(dbState)) {
entityStateManager.lambdaUpdate()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DEVICE.getIndex())
.eq(EntityStateDO::getEntityId, id)
.eq(EntityStateDO::getLeaseVersion, dbState.getLeaseVersion())
.set(EntityStateDO::getStateFlag, (byte) DeviceStatusEnum.OFFLINE.getIndex())
.set(EntityStateDO::getLeaseVersion, dbState.getLeaseVersion() + 1L)
.set(EntityStateDO::getExpireTime, LocalDateTime.now().plusSeconds(300))
.update();
}
FacadeDeviceBO device = deviceFacade.getById(id);
if (Objects.isNull(device)) {
log.debug("Device {} not found when handling offline expiry", id);
return;
}
if (Objects.isNull(device.getTenantId()) || device.getTenantId() <= 0) {
log.warn("Drop device offline alarm because tenantId could not be resolved, deviceId={}", id);
return;
}
String message = String.format("Device heartbeat timed out (last=%s); marked OFFLINE", lastStatus);
EntityAlarmDO entity = new EntityAlarmDO();
entity.setAlarmTargetTypeFlag(AlarmTargetTypeFlagEnum.DEVICE.getIndex());
entity.setEntityId(id);
entity.setDriverId(Objects.nonNull(device.getDriverId()) ? device.getDriverId() : 0L);
entity.setDeviceId(id);
entity.setPointId(0L);
entity.setRuleId(0L);
entity.setAlarmTypeFlag(AlarmTypeFlagEnum.OFFLINE.getIndex());
entity.setAlarmSourceFlag(AlarmSourceFlagEnum.STATE_TIMEOUT.getIndex());
// See handleDriverExpiry — heartbeat-timeout offline events default to P1.
entity.setAlarmLevelFlag(AlarmMessageLevelFlagEnum.P1.getIndex());
entity.setAlarmExt(JsonExt.builder()
.type("device-offline")
.content(message)
.version(1)
.build());
entity.setExpiredTime(0L);
entity.setConfirmFlag((byte) 0);
entity.setTenantId(device.getTenantId());
entityAlarmManager.save(entity);
DeviceAlarmDTO alarm = DeviceAlarmDTO.builder()
.driverId(device.getDriverId())
.tenantId(device.getTenantId())
.deviceId(id)
.status(null)
.statusName(null)
.message(message)
.alarmId(entity.getId())
.build();
alarmRuleTriggerService.processDeviceAlarm(alarm);
}
}
dc3-common/dc3-common-data/src/main/java/io/github/pnoker/common/data/rabbit/DriverTimeoutCheckReceiver.java
+183
View File
@@ -0,0 +1,183 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.data.rabbit;
import com.rabbitmq.client.Channel;
import io.github.pnoker.common.data.biz.alarm.AlarmRuleTriggerService;
import io.github.pnoker.common.data.dal.EntityAlarmManager;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityAlarmDO;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.entity.dto.DriverAlarmDTO;
import io.github.pnoker.common.entity.dto.DriverTimeoutCheckDTO;
import io.github.pnoker.common.entity.ext.JsonExt;
import io.github.pnoker.common.enums.AlarmMessageLevelFlagEnum;
import io.github.pnoker.common.enums.AlarmSourceFlagEnum;
import io.github.pnoker.common.enums.AlarmTargetTypeFlagEnum;
import io.github.pnoker.common.enums.AlarmTypeFlagEnum;
import io.github.pnoker.common.enums.DriverStatusEnum;
import io.github.pnoker.common.enums.EntityTypeFlagEnum;
import io.github.pnoker.common.utils.RabbitAckUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.util.Objects;
/**
* RabbitMQ receiver for driver timeout check messages.
* <p>
* Consumes messages dead-lettered from the 45s TTL delay queue and performs
* a secondary check against {@code dc3_entity_state}. Only marks the driver
* OFFLINE when the lease version, expiry, and online-family conditions all
* confirm the driver has truly stopped sending heartbeats.
*
* @author pnoker
* @version 2026.5.22
* @since 2026.5.22
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class DriverTimeoutCheckReceiver {
private static final int OFFLINE_RENEW_SECONDS = 300;
private final EntityStateManager entityStateManager;
private final EntityAlarmManager entityAlarmManager;
private final AlarmRuleTriggerService alarmRuleTriggerService;
@RabbitHandler
@RabbitListener(queues = "#{driverTimeoutCheckQueue.name}")
public void driverTimeoutCheck(Channel channel, Message message, DriverTimeoutCheckDTO dto) {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
if (Objects.isNull(dto) || Objects.isNull(dto.getDriverId()) || Objects.isNull(dto.getLeaseVersion())) {
RabbitAckUtil.reject(channel, deliveryTag);
return;
}
EntityStateDO state = entityStateManager.lambdaQuery()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DRIVER.getIndex())
.eq(EntityStateDO::getEntityId, dto.getDriverId())
.one();
// State row gone — nothing to do
if (Objects.isNull(state)) {
RabbitAckUtil.ack(channel, deliveryTag);
return;
}
// lease_version mismatched means a newer heartbeat arrived
if (!Objects.equals(state.getLeaseVersion(), dto.getLeaseVersion())) {
RabbitAckUtil.ack(channel, deliveryTag);
return;
}
// Not expired yet
if (state.getExpireTime().isAfter(LocalDateTime.now())) {
RabbitAckUtil.ack(channel, deliveryTag);
return;
}
// Already offline
byte offlineIndex = (byte) DriverStatusEnum.OFFLINE.getIndex();
if (Objects.equals(state.getStateFlag(), offlineIndex)) {
RabbitAckUtil.ack(channel, deliveryTag);
return;
}
// Online family check
boolean isOnline = state.getStateFlag() == DriverStatusEnum.ONLINE.getIndex()
|| state.getStateFlag() == DriverStatusEnum.MAINTAIN.getIndex();
if (!isOnline) {
RabbitAckUtil.ack(channel, deliveryTag);
return;
}
// Claim: atomically update to OFFLINE
long newVersion = state.getLeaseVersion() + 1L;
boolean claimed = entityStateManager.lambdaUpdate()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DRIVER.getIndex())
.eq(EntityStateDO::getEntityId, dto.getDriverId())
.eq(EntityStateDO::getLeaseVersion, state.getLeaseVersion())
.set(EntityStateDO::getLeaseVersion, newVersion)
.set(EntityStateDO::getStateFlag, offlineIndex)
.set(EntityStateDO::getLastStateFlag, state.getStateFlag())
.set(EntityStateDO::getExpireTime, LocalDateTime.now().plusSeconds(OFFLINE_RENEW_SECONDS))
.update();
if (!claimed) {
RabbitAckUtil.ack(channel, deliveryTag);
return;
}
// Write alarm
DriverStatusEnum prevStatus = DriverStatusEnum.ofIndex(state.getStateFlag());
String prevCode = Objects.nonNull(prevStatus) ? prevStatus.getCode() : "unknown";
String alarmMessage = String.format("Driver heartbeat timed out (last=%s); marked OFFLINE", prevCode);
EntityAlarmDO alarm = new EntityAlarmDO();
alarm.setAlarmTargetTypeFlag(AlarmTargetTypeFlagEnum.DRIVER.getIndex());
alarm.setEntityId(dto.getDriverId());
alarm.setDriverId(dto.getDriverId());
alarm.setDeviceId(0L);
alarm.setPointId(0L);
alarm.setRuleId(0L);
alarm.setRuleStateId(0L);
alarm.setAlarmTypeFlag(AlarmTypeFlagEnum.OFFLINE.getIndex());
alarm.setAlarmSourceFlag(AlarmSourceFlagEnum.STATE_TIMEOUT.getIndex());
alarm.setAlarmLevelFlag(AlarmMessageLevelFlagEnum.P1.getIndex());
alarm.setAlarmExt(JsonExt.builder().type("driver-offline").content(alarmMessage).version(1).build());
alarm.setExpiredTime(0L);
alarm.setConfirmFlag((byte) 0);
alarm.setTenantId(dto.getTenantId());
entityAlarmManager.save(alarm);
// Update lastAlarmId on state row
entityStateManager.lambdaUpdate()
.eq(EntityStateDO::getEntityTypeFlag, EntityTypeFlagEnum.DRIVER.getIndex())
.eq(EntityStateDO::getEntityId, dto.getDriverId())
.set(EntityStateDO::getLastAlarmId, alarm.getId())
.update();
// Trigger alarm rule pipeline
DriverAlarmDTO driverAlarm = DriverAlarmDTO.builder()
.tenantId(dto.getTenantId())
.driverId(dto.getDriverId())
.status(DriverStatusEnum.OFFLINE.getCode())
.statusName(DriverStatusEnum.OFFLINE.name())
.message(alarmMessage)
.alarmId(alarm.getId())
.build();
alarmRuleTriggerService.processDriverAlarm(driverAlarm);
log.info("Driver timeout check confirmed OFFLINE: driverId={}, tenantId={}, prevStatus={}",
dto.getDriverId(), dto.getTenantId(), prevCode);
RabbitAckUtil.ack(channel, deliveryTag);
} catch (Exception e) {
log.error("Driver timeout check failed, deliveryTag={}", deliveryTag, e);
RabbitAckUtil.nack(channel, deliveryTag, true);
}
}
}
dc3-common/dc3-common-data/src/test/java/io/github/pnoker/common/data/biz/impl/DeviceStateServiceImplTest.java
+24 -16
View File
@@ -19,7 +19,6 @@ package io.github.pnoker.common.data.biz.impl;
import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
import io.github.pnoker.common.data.biz.DeviceAlarmService;
import io.github.pnoker.common.data.cache.LocalCacheService;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.entity.dto.DeviceStateDTO;
@@ -36,7 +35,6 @@ import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -44,9 +42,6 @@ import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class DeviceStateServiceImplTest {
@Mock
private LocalCacheService localCacheService;
@Mock
private DeviceAlarmService deviceAlarmService;
@@ -77,8 +72,7 @@ class DeviceStateServiceImplTest {
}
@Test
void newDeviceCreatesDbRowWithCustomTtl() {
when(localCacheService.getKey(anyString())).thenReturn(null);
void newDeviceCreatesDbRowWithCustomTimeout() {
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(null);
@@ -92,11 +86,15 @@ class DeviceStateServiceImplTest {
EntityStateDO saved = captor.getValue();
assertThat(saved.getEntityTypeFlag()).isEqualTo((byte) EntityTypeFlagEnum.DEVICE.getIndex());
assertThat(saved.getEntityId()).isEqualTo(10L);
assertThat(saved.getDriverId()).isEqualTo(7L);
assertThat(saved.getParentEntityId()).isEqualTo(7L);
assertThat(saved.getTenantId()).isEqualTo(100L);
assertThat(saved.getLeaseVersion()).isEqualTo(1L);
assertThat(saved.getStateFlag()).isEqualTo((byte) DeviceStatusEnum.ONLINE.getIndex());
assertThat(saved.getTtlSeconds()).isEqualTo(25);
assertThat(saved.getTimeoutSeconds()).isEqualTo(25);
assertThat(saved.getLastStateFlag()).isEqualTo((byte) DeviceStatusEnum.OFFLINE.getIndex());
assertThat(saved.getLastHeartbeatTime()).isNotNull();
assertThat(saved.getLastAlarmId()).isEqualTo(0L);
assertThat(saved.getStateExt()).isNotNull();
}
@Test
@@ -105,8 +103,8 @@ class DeviceStateServiceImplTest {
existing.setEntityTypeFlag((byte) EntityTypeFlagEnum.DEVICE.getIndex());
existing.setEntityId(10L);
existing.setLeaseVersion(3L);
existing.setStateFlag((byte) DeviceStatusEnum.ONLINE.getIndex());
when(localCacheService.getKey(anyString())).thenReturn("online");
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(existing);
@@ -117,11 +115,11 @@ class DeviceStateServiceImplTest {
ArgumentCaptor<EntityStateDO> captor = ArgumentCaptor.forClass(EntityStateDO.class);
verify(entityStateManager).saveOrUpdate(captor.capture());
assertThat(captor.getValue().getLeaseVersion()).isEqualTo(4L);
assertThat(captor.getValue().getLastStateFlag()).isEqualTo((byte) DeviceStatusEnum.ONLINE.getIndex());
}
@Test
void nullDriverIdDefaultsToZero() {
when(localCacheService.getKey(anyString())).thenReturn(null);
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(null);
@@ -131,15 +129,20 @@ class DeviceStateServiceImplTest {
ArgumentCaptor<EntityStateDO> captor = ArgumentCaptor.forClass(EntityStateDO.class);
verify(entityStateManager).saveOrUpdate(captor.capture());
assertThat(captor.getValue().getDriverId()).isEqualTo(0L);
assertThat(captor.getValue().getParentEntityId()).isEqualTo(0L);
}
@Test
void statusFlipTriggersAlarm() {
when(localCacheService.getKey(anyString())).thenReturn("online");
EntityStateDO existing = new EntityStateDO();
existing.setEntityTypeFlag((byte) EntityTypeFlagEnum.DEVICE.getIndex());
existing.setEntityId(10L);
existing.setLeaseVersion(2L);
existing.setStateFlag((byte) DeviceStatusEnum.ONLINE.getIndex());
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(null);
when(queryWrapper.one()).thenReturn(existing);
when(entityStateManager.saveOrUpdate(any())).thenReturn(true);
service.heartbeat(heartbeat(10L, "offline", 7L, 100L, 25, TimeUnit.SECONDS));
@@ -149,10 +152,15 @@ class DeviceStateServiceImplTest {
@Test
void sameStatusNoAlarm() {
when(localCacheService.getKey(anyString())).thenReturn("online");
EntityStateDO existing = new EntityStateDO();
existing.setEntityTypeFlag((byte) EntityTypeFlagEnum.DEVICE.getIndex());
existing.setEntityId(10L);
existing.setLeaseVersion(2L);
existing.setStateFlag((byte) DeviceStatusEnum.ONLINE.getIndex());
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(null);
when(queryWrapper.one()).thenReturn(existing);
when(entityStateManager.saveOrUpdate(any())).thenReturn(true);
service.heartbeat(heartbeat(10L, "online", 7L, 100L, 25, TimeUnit.SECONDS));
dc3-common/dc3-common-data/src/test/java/io/github/pnoker/common/data/biz/impl/DriverStateServiceImplTest.java
+29 -17
View File
@@ -19,7 +19,6 @@ package io.github.pnoker.common.data.biz.impl;
import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
import io.github.pnoker.common.data.biz.DriverAlarmService;
import io.github.pnoker.common.data.cache.LocalCacheService;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.entity.dto.DriverStateDTO;
@@ -31,15 +30,13 @@ import org.mockito.ArgumentCaptor;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -47,9 +44,6 @@ import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class DriverStateServiceImplTest {
@Mock
private LocalCacheService localCacheService;
@Mock
private DriverAlarmService driverAlarmService;
@@ -59,6 +53,9 @@ class DriverStateServiceImplTest {
@Mock
private LambdaQueryChainWrapper<EntityStateDO> queryWrapper;
@Mock
private RabbitTemplate rabbitTemplate;
@InjectMocks
private DriverStateServiceImpl service;
@@ -74,7 +71,6 @@ class DriverStateServiceImplTest {
void nullDtoDoesNothing() {
service.heartbeat(null);
verify(entityStateManager, never()).saveOrUpdate(any());
verify(localCacheService, never()).setKey(anyString(), any(), anyLong(), any(TimeUnit.class));
}
@Test
@@ -87,7 +83,6 @@ class DriverStateServiceImplTest {
@Test
void newDriverCreatesDbRow() {
when(localCacheService.getKey(anyString())).thenReturn(null);
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(null);
@@ -101,14 +96,18 @@ class DriverStateServiceImplTest {
EntityStateDO saved = captor.getValue();
assertThat(saved.getEntityTypeFlag()).isEqualTo((byte) EntityTypeFlagEnum.DRIVER.getIndex());
assertThat(saved.getEntityId()).isEqualTo(1L);
assertThat(saved.getDriverId()).isEqualTo(1L);
assertThat(saved.getParentEntityId()).isEqualTo(1L);
assertThat(saved.getTenantId()).isEqualTo(100L);
assertThat(saved.getLeaseVersion()).isEqualTo(1L);
assertThat(saved.getStateFlag()).isEqualTo((byte) DriverStatusEnum.ONLINE.getIndex());
assertThat(saved.getTtlSeconds()).isEqualTo(45);
assertThat(saved.getTimeoutSeconds()).isEqualTo(45);
assertThat(saved.getExpireTime()).isAfter(LocalDateTime.now().plusSeconds(40));
assertThat(saved.getLastStateFlag()).isEqualTo((byte) DriverStatusEnum.OFFLINE.getIndex());
assertThat(saved.getLastHeartbeatTime()).isNotNull();
assertThat(saved.getLastAlarmId()).isEqualTo(0L);
assertThat(saved.getStateExt()).isNotNull();
verify(localCacheService).setKey(anyString(), eq("online"), eq(45L), eq(TimeUnit.SECONDS));
verify(rabbitTemplate).convertAndSend(anyString(), anyString(), any(Object.class));
}
@Test
@@ -117,8 +116,8 @@ class DriverStateServiceImplTest {
existing.setEntityTypeFlag((byte) EntityTypeFlagEnum.DRIVER.getIndex());
existing.setEntityId(1L);
existing.setLeaseVersion(5L);
existing.setStateFlag((byte) DriverStatusEnum.ONLINE.getIndex());
when(localCacheService.getKey(anyString())).thenReturn("online");
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(existing);
@@ -129,14 +128,22 @@ class DriverStateServiceImplTest {
ArgumentCaptor<EntityStateDO> captor = ArgumentCaptor.forClass(EntityStateDO.class);
verify(entityStateManager).saveOrUpdate(captor.capture());
assertThat(captor.getValue().getLeaseVersion()).isEqualTo(6L);
assertThat(captor.getValue().getLastStateFlag()).isEqualTo((byte) DriverStatusEnum.ONLINE.getIndex());
verify(rabbitTemplate).convertAndSend(anyString(), anyString(), any(Object.class));
}
@Test
void statusFlipFromOnlineToOfflineTriggersAlarm() {
when(localCacheService.getKey(anyString())).thenReturn("online");
EntityStateDO existing = new EntityStateDO();
existing.setEntityTypeFlag((byte) EntityTypeFlagEnum.DRIVER.getIndex());
existing.setEntityId(1L);
existing.setLeaseVersion(3L);
existing.setStateFlag((byte) DriverStatusEnum.ONLINE.getIndex());
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(null);
when(queryWrapper.one()).thenReturn(existing);
when(entityStateManager.saveOrUpdate(any())).thenReturn(true);
service.heartbeat(heartbeat(1L, "offline", 100L));
@@ -146,10 +153,15 @@ class DriverStateServiceImplTest {
@Test
void sameStatusNoFlipDoesNotTriggerAlarm() {
when(localCacheService.getKey(anyString())).thenReturn("online");
EntityStateDO existing = new EntityStateDO();
existing.setEntityTypeFlag((byte) EntityTypeFlagEnum.DRIVER.getIndex());
existing.setEntityId(1L);
existing.setLeaseVersion(3L);
existing.setStateFlag((byte) DriverStatusEnum.ONLINE.getIndex());
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.one()).thenReturn(null);
when(queryWrapper.one()).thenReturn(existing);
when(entityStateManager.saveOrUpdate(any())).thenReturn(true);
service.heartbeat(heartbeat(1L, "online", 100L));
dc3-common/dc3-common-data/src/test/java/io/github/pnoker/common/data/biz/impl/EntityStateExpiryScannerTest.java
+89 -103
View File
@@ -19,26 +19,30 @@ package io.github.pnoker.common.data.biz.impl;
import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
import com.baomidou.mybatisplus.extension.conditions.update.LambdaUpdateChainWrapper;
import com.rabbitmq.client.Channel;
import io.github.pnoker.common.data.biz.alarm.AlarmRuleTriggerService;
import io.github.pnoker.common.data.dal.EntityAlarmManager;
import io.github.pnoker.common.data.dal.EntityStateManager;
import io.github.pnoker.common.data.entity.model.EntityStateDO;
import io.github.pnoker.common.enums.DeviceStatusEnum;
import io.github.pnoker.common.enums.DriverStatusEnum;
import io.github.pnoker.common.enums.EntityTypeFlagEnum;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InOrder;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.inOrder;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
@@ -56,6 +60,12 @@ class EntityStateExpiryScannerTest {
@Mock
private AlarmRuleTriggerService alarmRuleTriggerService;
@Mock
private RabbitTemplate rabbitTemplate;
@Mock
private Channel channel;
@Mock
private LambdaQueryChainWrapper<EntityStateDO> queryWrapper;
@@ -65,23 +75,11 @@ class EntityStateExpiryScannerTest {
@InjectMocks
private EntityStateExpiryScanner scanner;
private EntityStateDO driverState(Long driverId, byte statusFlag, long leaseVersion, LocalDateTime expireTime) {
EntityStateDO state = new EntityStateDO();
state.setEntityTypeFlag((byte) EntityTypeFlagEnum.DRIVER.getIndex());
state.setEntityId(driverId);
state.setDriverId(driverId);
state.setStateFlag(statusFlag);
state.setLeaseVersion(leaseVersion);
state.setExpireTime(expireTime);
state.setTenantId(100L);
return state;
}
private EntityStateDO deviceState(Long deviceId, Long driverId, byte statusFlag, long leaseVersion, LocalDateTime expireTime) {
EntityStateDO state = new EntityStateDO();
state.setEntityTypeFlag((byte) EntityTypeFlagEnum.DEVICE.getIndex());
state.setEntityId(deviceId);
state.setDriverId(driverId);
state.setParentEntityId(driverId);
state.setStateFlag(statusFlag);
state.setLeaseVersion(leaseVersion);
state.setExpireTime(expireTime);
@@ -89,130 +87,118 @@ class EntityStateExpiryScannerTest {
return state;
}
@Test
void noExpiredRowsDoesNothing() {
private Message mockMessage(long deliveryTag) {
MessageProperties props = new MessageProperties();
props.setDeliveryTag(deliveryTag);
return new Message("tick".getBytes(), props);
}
private void stubDeviceQuery(List<EntityStateDO> results) {
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.eq(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.in(any(), any(Object[].class))).thenReturn(queryWrapper);
when(queryWrapper.lt(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.last(any())).thenReturn(queryWrapper);
when(queryWrapper.list()).thenReturn(Collections.emptyList());
when(queryWrapper.list()).thenReturn(results);
}
scanner.scanExpiredLeases();
private void stubClaimSuccess() {
when(entityStateManager.lambdaUpdate()).thenReturn(updateWrapper);
when(updateWrapper.eq(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.set(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.update()).thenReturn(true);
}
private void stubClaimFailure() {
when(entityStateManager.lambdaUpdate()).thenReturn(updateWrapper);
when(updateWrapper.eq(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.set(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.update()).thenReturn(false);
}
@Test
void noExpiredRowsDoesNothing() throws Exception {
stubDeviceQuery(Collections.emptyList());
scanner.onScanTick(channel, mockMessage(1L));
verify(rabbitTemplate).convertAndSend(anyString(), anyString(), anyString());
verifyNoInteractions(entityAlarmManager, alarmRuleTriggerService);
}
@Test
void alreadyOfflineDriverSkipsAlarm() {
EntityStateDO offline = driverState(1L,
(byte) DriverStatusEnum.OFFLINE.getIndex(),
void alreadyOfflineDeviceSkipsAlarm() throws Exception {
EntityStateDO offline = deviceState(10L, 7L,
(byte) DeviceStatusEnum.OFFLINE.getIndex(),
5L, LocalDateTime.now().minusSeconds(10));
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.lt(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.last(any())).thenReturn(queryWrapper);
when(queryWrapper.list()).thenReturn(List.of(offline));
stubDeviceQuery(List.of(offline));
// already offline path: lambdaUpdate for renewal
when(entityStateManager.lambdaUpdate()).thenReturn(updateWrapper);
when(updateWrapper.eq(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.set(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.update()).thenReturn(true);
scanner.scanExpiredLeases();
scanner.onScanTick(channel, mockMessage(1L));
verify(entityAlarmManager, never()).save(any());
verify(alarmRuleTriggerService, never()).processDriverAlarm(any());
verify(alarmRuleTriggerService, never()).processDeviceAlarm(any());
verify(rabbitTemplate).convertAndSend(anyString(), anyString(), anyString());
}
@Test
void onlineDriverExpiredWritesAlarmAndUpdatesState() {
EntityStateDO expired = driverState(1L,
(byte) DriverStatusEnum.ONLINE.getIndex(),
3L, LocalDateTime.now().minusSeconds(10));
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.lt(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.last(any())).thenReturn(queryWrapper);
when(queryWrapper.list()).thenReturn(List.of(expired));
// atomic claim
when(entityStateManager.lambdaUpdate()).thenReturn(updateWrapper);
when(updateWrapper.eq(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.set(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.update()).thenReturn(true);
when(entityAlarmManager.save(any())).thenReturn(true);
scanner.scanExpiredLeases();
verify(entityAlarmManager).save(any());
verify(alarmRuleTriggerService).processDriverAlarm(any());
}
@Test
void claimFailsWhenAnotherInstanceAlreadyProcessed() {
EntityStateDO expired = driverState(1L,
(byte) DriverStatusEnum.ONLINE.getIndex(),
3L, LocalDateTime.now().minusSeconds(10));
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.lt(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.last(any())).thenReturn(queryWrapper);
when(queryWrapper.list()).thenReturn(List.of(expired));
// atomic UPDATE returns false = another instance won
when(entityStateManager.lambdaUpdate()).thenReturn(updateWrapper);
when(updateWrapper.eq(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.set(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.update()).thenReturn(false);
scanner.scanExpiredLeases();
verify(entityAlarmManager, never()).save(any());
verify(alarmRuleTriggerService, never()).processDriverAlarm(any());
}
@Test
void onlineDeviceExpiredWritesAlarmAndUpdatesState() {
void onlineDeviceExpiredWritesAlarmAndUpdatesState() throws Exception {
EntityStateDO expired = deviceState(10L, 7L,
(byte) DeviceStatusEnum.ONLINE.getIndex(),
2L, LocalDateTime.now().minusSeconds(10));
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.lt(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.last(any())).thenReturn(queryWrapper);
when(queryWrapper.list()).thenReturn(List.of(expired));
when(entityStateManager.lambdaUpdate()).thenReturn(updateWrapper);
when(updateWrapper.eq(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.set(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.update()).thenReturn(true);
stubDeviceQuery(List.of(expired));
stubClaimSuccess();
when(entityAlarmManager.save(any())).thenReturn(true);
scanner.scanExpiredLeases();
scanner.onScanTick(channel, mockMessage(1L));
verify(entityAlarmManager).save(any());
verify(alarmRuleTriggerService).processDeviceAlarm(any());
verify(rabbitTemplate).convertAndSend(anyString(), anyString(), anyString());
}
@Test
void multipleExpiredRowsAllProcessed() {
EntityStateDO driver = driverState(1L,
(byte) DriverStatusEnum.ONLINE.getIndex(),
1L, LocalDateTime.now().minusSeconds(10));
EntityStateDO device = deviceState(10L, 1L,
void claimFailsWhenAnotherInstanceAlreadyProcessed() throws Exception {
EntityStateDO expired = deviceState(10L, 7L,
(byte) DeviceStatusEnum.ONLINE.getIndex(),
1L, LocalDateTime.now().minusSeconds(10));
3L, LocalDateTime.now().minusSeconds(10));
when(entityStateManager.lambdaQuery()).thenReturn(queryWrapper);
when(queryWrapper.lt(any(), any())).thenReturn(queryWrapper);
when(queryWrapper.last(any())).thenReturn(queryWrapper);
when(queryWrapper.list()).thenReturn(List.of(driver, device));
when(entityStateManager.lambdaUpdate()).thenReturn(updateWrapper);
when(updateWrapper.eq(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.set(any(), any())).thenReturn(updateWrapper);
when(updateWrapper.update()).thenReturn(true);
when(entityAlarmManager.save(any())).thenReturn(true);
stubDeviceQuery(List.of(expired));
stubClaimFailure();
scanner.scanExpiredLeases();
scanner.onScanTick(channel, mockMessage(1L));
InOrder inOrder = inOrder(alarmRuleTriggerService);
inOrder.verify(alarmRuleTriggerService).processDriverAlarm(any());
inOrder.verify(alarmRuleTriggerService).processDeviceAlarm(any());
verify(entityAlarmManager, never()).save(any());
verify(alarmRuleTriggerService, never()).processDeviceAlarm(any());
verify(rabbitTemplate).convertAndSend(anyString(), anyString(), anyString());
}
@Test
void scanTickPublishesNextTickOnSuccess() throws Exception {
stubDeviceQuery(Collections.emptyList());
scanner.onScanTick(channel, mockMessage(1L));
verify(rabbitTemplate).convertAndSend(
"dc3.e.state_timeout_delay",
"state.timeout.device.scan.tick",
"tick");
}
@Test
void scanTickNacksAndRequeuesOnFailure() throws Exception {
when(entityStateManager.lambdaQuery()).thenThrow(new RuntimeException("DB down"));
scanner.onScanTick(channel, mockMessage(1L));
verify(channel).basicNack(1L, false, true);
// next tick NOT published on failure
verify(rabbitTemplate, never()).convertAndSend(anyString(), anyString(), anyString());
}
}
dc3-common/dc3-common-model/src/main/java/io/github/pnoker/common/entity/dto/DriverTimeoutCheckDTO.java
+53
View File
@@ -0,0 +1,53 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.entity.dto;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.ToString;
import java.io.Serial;
import java.io.Serializable;
/**
* DTO published to the driver timeout delay queue when a driver heartbeat
* is processed. After the TTL expires the message is dead-lettered to the
* check queue where Data Center performs a lease-version comparison.
*
* @author pnoker
* @version 2026.5.22
* @since 2026.5.22
*/
@Getter
@Setter
@Builder
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class DriverTimeoutCheckDTO implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private Long driverId;
private Long leaseVersion;
private Long tenantId;
}