diff --git a/backend/iot-device/src/main/java/com/iot/device/collector/TuyaPulsarEventAdapter.java b/backend/iot-device/src/main/java/com/iot/device/collector/TuyaPulsarEventAdapter.java index eace7b2..d559e55 100644 --- a/backend/iot-device/src/main/java/com/iot/device/collector/TuyaPulsarEventAdapter.java +++ b/backend/iot-device/src/main/java/com/iot/device/collector/TuyaPulsarEventAdapter.java @@ -92,18 +92,74 @@ public class TuyaPulsarEventAdapter implements IPlatformEventAdapter { return events; } - /** 归一化事件类型（BizCode 与数字协议号双模式，与旧实现语义一致） */ + /** + * 归一化事件类型（BizCode 与数字协议号双模式） + * + *

修正：原实现把 {@code protocol == 4}（设备上下线）一律判为 ONLINE， + * 也把"含 status/online 字段"一律判为 ONLINE，导致离线推送被写成上线， + * 设备在线状态失真。现在一律按字段取值判定。

+ */ private String resolveType(Integer protocol, String bizCode, JsonNode d) { if ("deviceOnline".equals(bizCode)) return "ONLINE"; if ("deviceOffline".equals(bizCode)) return "OFFLINE"; if ("devicePropertyMessage".equals(bizCode) || "deviceReportMessage".equals(bizCode)) return "DP"; if (protocol != null) { - if (protocol == 4) return "ONLINE"; + if (protocol == 4) { + // 上下线：按 payload 里的状态值判定，不能一律当上线 + return resolveOnlineByPayload(d) ? "ONLINE" : "OFFLINE"; + } if (protocol == 8 || protocol == 9) return "DP"; } // 数字 protocol 老结构里 4 之外，data 里带 status/online 的也算上下线 - if (d.has("status") || d.has("online")) return "ONLINE"; + if (d.has("status") || d.has("online")) { + return resolveOnlineByPayload(d) ? "ONLINE" : "OFFLINE"; + } return "OTHER"; } + + /** + * 从推送负载里解析设备是否在线 + * + *

兼容三种取值：布尔 true/false、数字 1/0、字符串 "online"/"offline"（或 "1"/"0"）。

+ */ + private boolean resolveOnlineByPayload(JsonNode d) { + if (d == null) { + return false; + } + JsonNode online = d.get("online"); + if (online != null && !online.isNull()) { + if (online.isBoolean()) { + return online.asBoolean(); + } + if (online.isNumber()) { + return online.asInt() != 0; + } + String text = online.asText(); + if ("online".equalsIgnoreCase(text) || "1".equals(text)) { + return true; + } + if ("offline".equalsIgnoreCase(text) || "0".equals(text)) { + return false; + } + } + JsonNode status = d.get("status"); + if (status != null && !status.isNull()) { + if (status.isBoolean()) { + return status.asBoolean(); + } + if (status.isNumber()) { + return status.asInt() != 0; + } + String text = status.asText(); + if ("online".equalsIgnoreCase(text) || "1".equals(text)) { + return true; + } + if ("offline".equalsIgnoreCase(text) || "0".equals(text)) { + return false; + } + } + // 取不到明确状态时，按"无状态证据"处理为离线，避免把离线误判成在线 + return false; + } } diff --git a/backend/iot-device/src/main/java/com/iot/device/service/TuyaMessageSubscribeService.java b/backend/iot-device/src/main/java/com/iot/device/service/TuyaMessageSubscribeService.java index 3bcb50b..807b33a 100644 --- a/backend/iot-device/src/main/java/com/iot/device/service/TuyaMessageSubscribeService.java +++ b/backend/iot-device/src/main/java/com/iot/device/service/TuyaMessageSubscribeService.java @@ -7,9 +7,11 @@ import com.iot.device.collector.PlatformEventInput; import com.iot.device.collector.TuyaPulsarEventAdapter; import com.iot.device.dto.DeviceEvent; import com.iot.device.entity.Device; +import com.iot.device.entity.DeviceState; import com.iot.device.entity.PlatformConfig; import com.iot.device.event.DeviceEventPublisher; import com.iot.device.mapper.DeviceMapper; +import com.iot.device.mapper.DeviceStateMapper; import com.tuya.open.sdk.mq.AESBase64Utils; import com.tuya.open.sdk.mq.MqConfigs; import com.tuya.open.sdk.mq.MqConsumer; @@ -63,14 +65,32 @@ public class TuyaMessageSubscribeService implements SmartLifecycle { private final TuyaCloudService tuyaCloudService; private final DeviceMapper deviceMapper; + private final DeviceStateMapper deviceStateMapper; private final DeviceEventPublisher deviceEventPublisher; private final ObjectMapper objectMapper; private final DeviceActivityService deviceActivityService; private final TuyaPulsarEventAdapter tuyaPulsarEventAdapter; private volatile boolean running = false; + /** + * 是否已收到过消息（真正"连上了"的唯一可靠证据） + * + *

涂鸦 SDK 的 MqConsumer 没有暴露连接状态接口，且 {@code start()} 是阻塞式的 + * （内部 subscribe + receive），因此"调用了 start()"既不能证明连上、也不能阻塞主循环。 + * 这里用"是否收到过消息"作为连接成功的判据。

+ */ private volatile boolean connected = false; + /** 是否已发起连接（start() 已调用） */ + private volatile boolean connecting = false; + /** 累计收到的消息条数 */ + private volatile long messageCount = 0L; + /** 最近一次发起连接的时间 */ + private volatile long lastConnectAt = 0L; + /** 最近一次连接错误 */ + private volatile String lastError = ""; private volatile String currentAccessKey; + /** 当前实际使用的订阅服务器地址（按区域解析） */ + private volatile String currentServerUrl = ""; private volatile long lastMessageAt = 0L; private volatile String lastMessageSummary = "尚未收到消息"; private Thread subscribeThread; @@ -119,6 +139,7 @@ public class TuyaMessageSubscribeService implements SmartLifecycle { String region = StringUtils.hasText(config.getApiRegion()) ? config.getApiRegion().toLowerCase() : "cn"; String serverUrl = resolveServerUrl(region); + currentServerUrl = serverUrl; currentAccessKey = config.getAccessSecret(); MqConsumer consumer = MqConsumer.build() @@ -129,23 +150,67 @@ public class TuyaMessageSubscribeService implements SmartLifecycle { log.info("涂鸦消息订阅启动中: serverUrl={}, accessId={}, region={}", serverUrl, config.getAccessId(), region); - consumer.start(); - // start() 正常返回：连接已建立，SDK 在内部线程接收消息，这里保活等待 - connected = true; - log.info("涂鸦消息订阅已连接，开始接收设备状态推送（上下线/DP上报）"); + + // SDK 的 start() 是阻塞式的（内部完成 subscribe 后一直 receive）， + // 必须在独立线程里调用： + // 1) 主循环才能感知"消费者线程已退出"并触发重连； + // 2) 否则连接一旦在 start() 内抛异常就没有任何重试机会。 + final MqConsumer finalConsumer = consumer; + Thread consumerThread = new Thread(() -> { + try { + finalConsumer.start(); + } catch (Exception e) { + lastError = e.getClass().getSimpleName() + ": " + e.getMessage(); + log.warn("涂鸦消息订阅消费者异常退出: {}", e.getMessage()); + } finally { + connecting = false; + } + }, "tuya-mq-consumer"); + consumerThread.setDaemon(true); + consumerThread.start(); + connecting = true; + connected = false; + lastConnectAt = System.currentTimeMillis(); + lastError = ""; + + // 给 SDK 一点时间完成鉴权与订阅：若立刻失败（认证/订阅不存在）， + // 消费者线程会很快退出，这里据此提前失败并进入退避重试。 + Thread.sleep(3000L); + if (!consumerThread.isAlive()) { + connecting = false; + throw new IllegalStateException("涂鸦消息订阅消费者启动后立即退出，请检查云项目" + + "「消息订阅」是否已开通（且已开通本项目产品的消息类型）"); + } + log.info("涂鸦消息订阅已发起连接（SDK 阻塞式接收，等待首条消息确认连通性）"); + + // 保活：监控消费者线程，退出即重连 while (running) { Thread.sleep(5000L); + if (!consumerThread.isAlive()) { + connecting = false; + connected = false; + log.warn("涂鸦消息订阅消费者线程已退出，触发重连"); + break; + } } - connected = false; - return; + if (!running) { + return; + } + sleepQuietly(backoffMs); + backoffMs = Math.min(backoffMs * 2, 600_000L); + continue; } catch (InterruptedException ie) { Thread.currentThread().interrupt(); connected = false; + connecting = false; return; } catch (Exception e) { connected = false; + connecting = false; + lastError = e.getClass().getSimpleName() + ": " + e.getMessage(); log.warn("涂鸦消息订阅连接失败/中断: {}，{}秒后重试" - + "（若持续失败请到涂鸦开发者平台云项目的「消息订阅」页面确认已开启）", + + "（若持续失败请到涂鸦开发者平台云项目的「消息订阅」页面确认已开启，" + + "并确认已订阅本项目产品的消息类型）", e.getMessage(), backoffMs / 1000); sleepQuietly(backoffMs); backoffMs = Math.min(backoffMs * 2, 600_000L); @@ -158,6 +223,8 @@ public class TuyaMessageSubscribeService implements SmartLifecycle { private void onMessageArrived(org.apache.pulsar.client.api.Message message) { try { lastMessageAt = System.currentTimeMillis(); + messageCount++; + connected = true; // 收到消息 = 连接与订阅均已生效（唯一可靠证据） String payload = new String(message.getData(), StandardCharsets.UTF_8); JsonNode root = objectMapper.readTree(payload); // 涂鸦推送存在两种标识：老协议用数字 protocol（4=上下线, 8/9=DP 上报）， @@ -206,6 +273,11 @@ public class TuyaMessageSubscribeService implements SmartLifecycle { if ("online".equals(ev.getType()) || "offline".equals(ev.getType())) { syncDeviceOnlineStatus(ev.getDeviceId(), "online".equals(ev.getType())); } + // 属性上报落 device_state：子设备（网关下挂）没有独立在线通道， + // 但只要有属性上报就说明它通电且链路正常 —— 这是判断子设备在线的权威证据。 + if ("state".equals(ev.getType()) && ev.getDps() != null && !ev.getDps().isEmpty()) { + syncDeviceStateFromDps(ev.getDeviceId(), ev.getDps()); + } } lastMessageSummary = effectiveBizCode + " → " + events.size() + " 事件"; } @@ -311,14 +383,89 @@ public class TuyaMessageSubscribeService implements SmartLifecycle { } } + /** + * 把涂鸦属性上报写入 device_state（属性级当前状态）。 + * + *

device_state 此前只有中控屏上报一条写入路径，导致"子设备最近是否有活动"这类 + * 在线证据无处可查（表恒为空，看起来像订阅没通）。这里把云端属性上报也落到同一张表： + * 网关下挂的子设备虽无独立在线通道，但只要有属性上报，就说明它通电且链路正常 —— + * 这正是判定子设备在线的权威证据（强于"父网关在线"的推断）。

+ */ + private void syncDeviceStateFromDps(String devId, Map dps) { + if (!StringUtils.hasText(devId) || dps == null || dps.isEmpty()) { + return; + } + Device device = deviceMapper.selectByDeviceId(devId); + if (device == null) { + log.debug("涂鸦推送的设备未入库，忽略属性状态落库: {}", devId); + return; + } + java.time.LocalDateTime now = java.time.LocalDateTime.now(); + for (Map.Entry entry : dps.entrySet()) { + String code = entry.getKey(); + if (!StringUtils.hasText(code)) { + continue; + } + DeviceState existing = deviceStateMapper.selectOne( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .eq(DeviceState::getDeviceId, devId) + .eq(DeviceState::getPropertyCode, code) + .last("LIMIT 1")); + DeviceState state = existing != null ? existing : new DeviceState(); + state.setDeviceId(devId); + state.setPropertyCode(code); + state.setPropertyValue(entry.getValue() == null ? null : String.valueOf(entry.getValue())); + state.setQuality("GOOD"); + state.setReportedAt(now); + state.setReceivedAt(now); + if (state.getId() == null) { + deviceStateMapper.insert(state); + } else { + deviceStateMapper.updateById(state); + } + } + // 顺带把最新 DP 写入 device.capabilities，供在线判定与详情展示使用 + try { + com.fasterxml.jackson.databind.JsonNode capsNode = device.getCapabilities(); + java.util.Map capsMap = new java.util.LinkedHashMap<>(); + if (capsNode != null && capsNode.isObject()) { + capsMap.putAll(objectMapper.convertValue(capsNode, java.util.Map.class)); + } + capsMap.put("dps", dps); + Device update = new Device(); + update.setId(device.getId()); + update.setCapabilities(objectMapper.valueToTree(capsMap)); + deviceMapper.updateById(update); + } catch (Exception e) { + log.debug("写入 device.capabilities 失败: {}", e.getMessage()); + } + } + // ==================== 状态查询（供管理端/排查） ==================== - /** 订阅连接状态 */ + /** + * 订阅连接状态 + * + *

字段说明：

+ *
    + *
  • {@code running}：订阅线程是否在运行；
    • + *
  • {@code connecting}：是否已发起连接（SDK 已调用 start()）；
    • + *
  • {@code connected}：是否收到过消息 —— SDK 没有暴露连接状态接口， + * 这是唯一可靠的"真的连上了"证据；
    • + *
  • {@code messageCount}：累计收到的消息条数；
    • + *
  • {@code lastError}：最近一次连接错误（便于定位认证/订阅问题）。
    • + *
+ */ public Map getStatus() { Map m = new LinkedHashMap<>(); m.put("running", running); + m.put("connecting", connecting); m.put("connected", connected); - m.put("serverUrl", MqConfigs.CN_SERVER_URL); + m.put("messageCount", messageCount); + m.put("lastConnectAt", lastConnectAt); + m.put("lastError", lastError); + m.put("serverUrl", StringUtils.hasText(currentServerUrl) + ? currentServerUrl : MqConfigs.CN_SERVER_URL); m.put("lastMessageAt", lastMessageAt); m.put("lastMessageSummary", lastMessageSummary); return m; diff --git a/backend/iot-device/src/test/java/com/iot/device/collector/TuyaPulsarEventAdapterTest.java b/backend/iot-device/src/test/java/com/iot/device/collector/TuyaPulsarEventAdapterTest.java new file mode 100644 index 0000000..1675d06 --- /dev/null +++ b/backend/iot-device/src/test/java/com/iot/device/collector/TuyaPulsarEventAdapterTest.java @@ -0,0 +1,77 @@ +package com.iot.device.collector; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.iot.device.dto.DeviceEvent; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * 涂鸦推送事件归一化单测 + * + *

重点：protocol=4（设备上下线）必须按 payload 的 status 取值判定， + * 不能一律当成上线 —— 否则离线推送会把设备状态写成在线。

+ */ +@DisplayName("涂鸦推送事件归一化") +class TuyaPulsarEventAdapterTest { + + private final ObjectMapper mapper = new ObjectMapper(); + private final TuyaPulsarEventAdapter adapter = new TuyaPulsarEventAdapter(mapper); + + private List normalize(String payloadJson, Integer protocol, String bizCode) throws Exception { + PlatformEventInput input = new PlatformEventInput(); + input.setPlatform(DevicePlatform.TUYA.code()); + input.setPayload(mapper.readTree(payloadJson)); + input.setBizCode(bizCode); + input.setProtocol(protocol); + input.setRawJson(payloadJson); + return adapter.normalize(input); + } + + @Test + @DisplayName("protocol=4 且 status=0 判为离线") + void shouldResolveOfflineWhenStatusZero() throws Exception { + List events = normalize("{\"devId\":\"dev_1\",\"status\":0}", 4, null); + assertEquals(1, events.size()); + assertEquals("offline", events.get(0).getType()); + } + + @Test + @DisplayName("protocol=4 且 status=1 判为在线") + void shouldResolveOnlineWhenStatusOne() throws Exception { + List events = normalize("{\"devId\":\"dev_1\",\"status\":1}", 4, null); + assertEquals(1, events.size()); + assertEquals("online", events.get(0).getType()); + } + + @Test + @DisplayName("protocol=4 且 online=false 判为离线") + void shouldResolveOfflineWhenOnlineFalse() throws Exception { + List events = normalize("{\"devId\":\"dev_1\",\"online\":false}", 4, null); + assertEquals(1, events.size()); + assertEquals("offline", events.get(0).getType()); + } + + @Test + @DisplayName("新版 bizCode=deviceOffline 判为离线") + void shouldResolveOfflineByBizCode() throws Exception { + List events = normalize("{\"devId\":\"dev_1\",\"status\":0}", 1001, "deviceOffline"); + assertEquals(1, events.size()); + assertEquals("offline", events.get(0).getType()); + } + + @Test + @DisplayName("属性上报解析出 dps") + void shouldParseDpPayload() throws Exception { + List events = normalize( + "{\"devId\":\"dev_1\",\"dps\":{\"switch_1\":true}}", 8, null); + assertEquals(1, events.size()); + DeviceEvent event = events.get(0); + assertEquals("state", event.getType()); + assertTrue(event.getDps() != null && event.getDps().containsKey("switch_1")); + } +}