- * 监听 IoT 模块发布的设备状态变更事件（上线/离线） - *
- * 调用链路： - * - *
- * IoT 设备连接/断开 - * ↓ - * IotDeviceServiceImpl.updateDeviceState() - * ↓ - * RocketMQ Topic: integration-device-status - * ↓ - * BadgeDeviceStatusEventHandler.onMessage() - * - 幂等性检查 - * - 判断是否为工牌设备 - * ↓ - * badgeDeviceStatusService.updateBadgeOnlineStatus() - * - 创建/更新 Redis 状态记录 - *- * - * @author AI - */ -@Slf4j -@Component -@RocketMQMessageListener( - topic = "integration-device-status", - consumerGroup = "ops-badge-status-consumer", - consumeMode = ConsumeMode.CONCURRENTLY, - selectorExpression = "*", - accessKey = "${rocketmq.consumer.access-key:}", - secretKey = "${rocketmq.consumer.secret-key:}" -) -public class BadgeDeviceStatusEventHandler implements RocketMQListener
- * 通过查询 ops_area_device_relation 表判断设备是否为 BADGE 类型
- */
- private boolean isBadgeDevice(Long deviceId) {
- if (deviceId == null) {
- return false;
- }
-
- try {
- // 通过区域设备关系判断是否为工牌设备
- OpsAreaDeviceRelationDO relation = areaDeviceService.getByDeviceId(deviceId);
- return relation != null && "BADGE".equals(relation.getRelationType());
- } catch (Exception e) {
- log.warn("[BadgeDeviceStatusEventHandler] 查询设备类型失败: deviceId={}", deviceId, e);
- return false;
- }
- }
-
- /**
- * 获取设备所属区域ID
- */
- private Long getAreaIdByDeviceId(Long deviceId) {
- if (deviceId == null) {
- return null;
- }
-
- try {
- OpsAreaDeviceRelationDO relation = areaDeviceService.getByDeviceId(deviceId);
- if (relation != null && "BADGE".equals(relation.getRelationType())) {
- return relation.getAreaId();
- }
- return null;
- } catch (Exception e) {
- log.warn("[BadgeDeviceStatusEventHandler] 获取设备区域失败: deviceId={}", deviceId, e);
- return null;
- }
- }
-}
+package com.viewsh.module.ops.environment.integration.consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.viewsh.framework.tenant.core.context.TenantContextHolder;
+import com.viewsh.framework.tenant.core.util.TenantUtils;
+import com.viewsh.module.ops.dal.dataobject.area.OpsAreaDeviceRelationDO;
+import com.viewsh.module.ops.enums.BadgeDeviceStatusEnum;
+import com.viewsh.module.ops.environment.integration.dto.IotDeviceStatusChangedEventDTO;
+import com.viewsh.module.ops.environment.service.badge.BadgeDeviceStatusService;
+import com.viewsh.module.ops.infrastructure.redis.OpsRedisKeyBuilder;
+import com.viewsh.module.ops.service.area.AreaDeviceService;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+@RocketMQMessageListener(
+ topic = "integration-device-status",
+ consumerGroup = "ops-badge-status-consumer",
+ consumeMode = ConsumeMode.CONCURRENTLY,
+ selectorExpression = "*",
+ accessKey = "${rocketmq.consumer.access-key:}",
+ secretKey = "${rocketmq.consumer.secret-key:}"
+)
+public class BadgeDeviceStatusEventHandler implements RocketMQListener
- * 订阅 IoT 模块发布的保洁工单到岗事件
- *
- * 调用链路：
- *
- * 设计说明：
- * - Handler 只负责消息接收和状态转换
- * - 业务日志、设备缓存更新由 CleanOrderEventListener 在事务提交后处理
- * - BadgeDeviceStatusServiceImpl 在 BEFORE_COMMIT 阶段跳过 ARRIVED（由 Listener 处理完整信息）
- *
- * @author AI
- */
-@Slf4j
-@Component
-@RocketMQMessageListener(
- topic = "ops-order-arrive",
- consumerGroup = "ops-clean-order-arrive-group",
- consumeMode = ConsumeMode.CONCURRENTLY,
- selectorExpression = "*",
- accessKey = "${rocketmq.consumer.access-key:}",
- secretKey = "${rocketmq.consumer.secret-key:}"
-)
-public class CleanOrderArriveEventHandler implements RocketMQListener
- * 订阅 IoT 模块发布的保洁工单完成事件
- *
- * 调用链路：
- *
- *
- * 设计说明：
- * - Handler 只负责消息接收和状态转换
- * - 设备状态由 BadgeDeviceStatusEventListener 在 BEFORE_COMMIT 阶段处理
- * - 业务日志、自动调度由 CleanOrderEventListener 在 AFTER_COMMIT 阶段处理
- *
- * @author AI
- */
-@Slf4j
-@Component
-@RocketMQMessageListener(
- topic = "ops-order-complete",
- consumerGroup = "ops-clean-order-complete-group",
- consumeMode = ConsumeMode.CONCURRENTLY,
- selectorExpression = "*",
- accessKey = "${rocketmq.consumer.access-key:}",
- secretKey = "${rocketmq.consumer.secret-key:}"
-)
-public class CleanOrderCompleteEventHandler implements RocketMQListener
- * 订阅 IoT 模块发布的工单确认事件（如：工牌按键确认）
- *
- * 调用链路：
- *
- * 设计说明：
- * - Handler 只负责消息接收和状态转换
- * - 日志记录和通知发送由 CleanOrderEventListener 在事务提交后处理
- * - 设备状态由 BadgeDeviceStatusServiceImpl 在 BEFORE_COMMIT 阶段同步
- *
- * @author AI
- */
@Slf4j
@Component
@RocketMQMessageListener(
@@ -60,7 +33,6 @@ import java.util.concurrent.TimeUnit;
)
public class CleanOrderConfirmEventHandler implements RocketMQListener
- * IoT 发布 RocketMQ 消息 (ops-order-arrive)
- * ↓
- * CleanOrderArriveEventHandler.onMessage()
- * - 幂等性检查
- * - 状态检查
- * ↓
- * orderLifecycleManager.transition(ARRIVED) [@Transactional]
- * - 更新工单状态
- * - 发布 OrderStateChangedEvent
- * ↓
- * 事务提交
- * ↓
- * CleanOrderEventListener.handleArrived() [@Async, AFTER_COMMIT]
- * - 记录到岗时间到扩展表
- * - 更新设备工单关联信息（areaId, beaconMac）
- * - 记录业务日志
- *
- *
- * IoT 发布 RocketMQ 消息 (ops-order-complete)
- * ↓
- * CleanOrderCompleteEventHandler.onMessage()
- * - 幂等性检查
- * - 状态检查
- * ↓
- * orderLifecycleManager.transition(COMPLETED) [@Transactional]
- * - 更新工单状态
- * - 发布 OrderStateChangedEvent
- * ↓
- * 事务提交
- * ↓
- * BadgeDeviceStatusEventListener.onOrderStateChanged() [BEFORE_COMMIT]
- * - 清除设备工单关联
- * - 检查等待任务，决定设备状态 (BUSY/IDLE)
- * ↓
- * CleanOrderEventListener.handleCompleted() [AFTER_COMMIT]
- * - 记录完成时间到扩展表
- * - 记录业务日志
- * CleanOrderEventListener.onOrderCompleted() [AFTER_COMMIT]
- * - 自动调度下一个任务
- * - 发送完成通知
- *
- *
- * IoT 发布 RocketMQ 消息 (ops-order-confirm)
- * ↓
- * CleanOrderConfirmEventHandler.onMessage()
- * - 幂等性检查
- * - 状态检查
- * ↓
- * orderLifecycleManager.transition(CONFIRMED) [@Transactional]
- * - 更新工单状态
- * - 发布 OrderStateChangedEvent
- * ↓
- * 事务提交
- * ↓
- * CleanOrderEventListener.onOrderStateChanged() [@TransactionalEventListener(AFTER_COMMIT)]
- * - 记录业务日志
- * - 发送 TTS "工单已确认，请前往作业区域开始作业"
- *
- *