4.10 HMS功能

• sample代码架构
• sample代码实现
• 开发注意事项

4.10.1 sample代码架构

HMS告警功能目前只在机场支持，遥控器Pilot2上报的HMS告警信息是无效的。

4.10.2 sample代码实现

所有mqtt消息都会经过InboundMessageRouter#determineTargetChannels方法，在该方法中找到通过DeviceTopicEnum#find方法找到消息对应的主题，如果是 thing/product/{device_sn}/events 主题的消息会被选择到ChannelName.INBOUND_EVENTS通道中。

public class InboundMessageRouter extends AbstractMessageRouter {

    @Override
    @Router(inputChannel = ChannelName.INBOUND)
    protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
        // 获取消息头
        MessageHeaders headers = message.getHeaders();
        String topic = headers.get(MqttHeaders.RECEIVED_TOPIC).toString();
        byte[] payload = (byte[])message.getPayload();
        log.debug("received topic :{} \t payload :{}", topic, new String(payload));
        // 根据消息topic不同，将消息转发到不同的channel中。
        //  thing/product/{device_sn}/events 主题消息会被转发到ChannelName.INBOUND_EVENTS通道中。
        DeviceTopicEnum topicEnum = DeviceTopicEnum.find(topic);
        MessageChannel bean = (MessageChannel) SpringBeanUtils.getBean(topicEnum.getBeanName());
        return Collections.singleton(bean);
    }
}

上报固件版本号的消息被转发到ChannelName.INBOUND_EVENTS中后，通过

public IntegrationFlow eventsMethodRouterFlow() {
    return IntegrationFlows
        .from(ChannelName.INBOUND_EVENTS)
        .transform(Message.class, source -> {
            try {
                TopicEventsRequest data = Common.getObjectMapper().readValue((byte[]) source.getPayload(), TopicEventsRequest.class);
                String topic = String.valueOf(source.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
                // 获取HMS上报设备的sn，并将sn填充进消息实体
                return data.setFrom(topic.substring((THING_MODEL_PRE + PRODUCT).length(), topic.indexOf(EVENTS_SUF)))
                    .setData(Common.getObjectMapper().convertValue(data.getData(), EventsMethodEnum.find(data.getMethod()).getClassType()));
            } catch (IOException e) {
                throw new CloudSDKException(e);
            }
        }, null)
        .<TopicEventsRequest, EventsMethodEnum>route(
            // 通过上报消息的method属性确定上报消息需要处理的通道
            // method:"hms"的消息被传递给ChannelName.INBOUND_EVENTS_HMS通道
            response -> EventsMethodEnum.find(response.getMethod()),
                mapping -> Arrays.stream(EventsMethodEnum.values()).forEach(
                    methodEnum -> mapping.channelMapping(methodEnum, methodEnum.getChannelName())))
        .get();
}

ChannelName.INBOUND_EVENTS_HMS通道消息会被AbstractHmsService#hms方法进行处理，在sdk包中默认实现抛出异常。

@ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_HMS)
public void hms(TopicEventsRequest<Hms> response, MessageHeaders headers) {
    throw new UnsupportedOperationException("hms not implemented");
}

开发者需要自定义实现逻辑，可以定义AbstractHmsService类的实现类，实现AbstractHmsService#hms方法。示例代码的实现为DeviceHmsServiceImpl#hms：

@Override
public void hms(TopicEventsRequest<Hms> response, MessageHeaders headers) {
    // 获取到消息上报的设备sn
    String sn = response.getFrom();
    // 将消息构造为告警信息的实体
    DeviceHmsEntity entity = DeviceHmsEntity.builder()
        .bid(response.getBid())
        .tid(response.getTid())
        .createTime(response.getTimestamp())
        .updateTime(0L)
        .sn(sn)
        .build();
    // 查询所有没有查看过的告警信息，所有告警信息都存在Redis中
    // 如果在前端查看过了，会从Redis中进行删除
    Set<String> hmsMap = deviceRedisService.getAllHmsKeys(sn);

    List<DeviceHmsDTO> unReadList = new ArrayList<>();
    response.getData().getList()
        .forEach(hmsReceiver -> {
            final DeviceHmsEntity hms = entity.clone();
            this.fillEntity(hms, hmsReceiver);
            // 如果再次上报的HMS消息，在未读的HMS列表中，则不再重复添加
            if (hmsMap.contains(hms.getHmsKey())) {
                return;
            }
            // 根据hms.json文件中的内容，将上报的hms消息进行填充转义
            this.fillMessage(hms, hmsReceiver.getArgs());
            unReadList.add(entity2Dto(hms));
            mapper.insert(hms);
        });

    // 如果告警信息为空，直接返回
    if (unReadList.isEmpty()) {
       return;
    }
    // 刷新Redis中缓存的HMS告警信息
    deviceRedisService.addEndHmsKeys(sn, unReadList.stream().map(DeviceHmsDTO::getKey).toArray(String[]::new));
    // 检查设备是否在线，如果设备不在线，就不需要向前端展示
    Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(sn);
    if (deviceOpt.isEmpty()) {
        return;
    }
    // 通过websocket协议通知，HMS信息进行更新
    sendMessageService.sendBatch(deviceOpt.get().getWorkspaceId(), UserTypeEnum.WEB.getVal(),
        BizCodeEnum.DEVICE_HMS.getCode(), TelemetryDTO.<List<DeviceHmsDTO>>builder().sn(sn).host(unReadList).build());
}

4.10.3 开发注意事项

1. HMS是全量上报，如果当前上报的HMS信息中没有上一次的告警，则说明告警已经解除。