MiddlePlatform/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java

package com.fastbee.service;


import cn.hutool.core.io.IoUtil;
import com.alibaba.fastjson2.JSON;
import com.fastbee.common.CommandConstant;
import com.fastbee.common.NettyCacheKey;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.dto.DeviceErrorPushDTO;
import com.fastbee.dto.DeviceErrorPushDetail;
import com.fastbee.dto.NettyDevice;
import com.fastbee.iot.domain.DeviceRunningStatusHistory;
import com.fastbee.entity.ErrorIdentificationStandard;
import com.fastbee.enums.CompareEnum;
import com.fastbee.enums.RespondCode;
import com.fastbee.iot.domain.DeviceStatusRecordDTO;
import com.fastbee.manager.DeviceSessionManager;
import com.fastbee.mapper.ErrorIdentificationStandardMapper;
import com.fastbee.memory.GlobalMemory;
import com.fastbee.netty.handler.DeviceConnectHandler;
import com.fastbee.utils.CommandUtils;
import com.fastbee.utils.CompareUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class DeviceThreadService {

    @Autowired
    private ResourceLoader resourceLoader;

    @Autowired
    private GlobalMemory globalMemory;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private DeviceSessionManager sessionManager;

    @Autowired
    private ErrorIdentificationStandardMapper standardMapper;


    public void deviceErrorMonitor() throws InterruptedException, IOException {
        initStandard();

        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleWithFixedDelay(() -> {
            try {
                String key = NettyCacheKey.deviceRunningStatusKey;
                Collection<String> keys = redisCache.getListKeyByPrefix(key);
                if (!CollectionUtils.isEmpty(keys)) {
                    keys.forEach(k -> {
                        DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(k);
                        if (recordDTO != null && !CollectionUtils.isEmpty(recordDTO.getHistories())) {
                            recordDTO.getHistories().sort(Comparator.comparing(DeviceRunningStatusHistory::getCreateTime).reversed());
                            DeviceRunningStatusHistory history = recordDTO.getHistories().get(0);
                            List<DeviceErrorPushDetail> compared = compareErrorStandard(history, globalMemory.standard);

                            String[] parts = k.split(":");
                            String deviceId = parts[1];

                            DeviceErrorPushDTO pushDTO = new DeviceErrorPushDTO();
                            pushDTO.setDeviceId(deviceId);
                            pushDTO.setTime(System.currentTimeMillis());
                            pushDTO.setDetails(compared);
                            pushDTO.setEvent(RespondCode.device_error_push);
                            //todo push 推送到前端
                            pushErrorMessage(deviceId, JSON.toJSONString(pushDTO));
                        }
                    });
                }
            } catch (Exception e) {
                log.error("执行监测错误线程错误：{}", e.getMessage());
            }
        }, 0, 2, TimeUnit.MINUTES);
    }

    public void initStandard() throws IOException {
        if (globalMemory.standard == null) {
            List<ErrorIdentificationStandard> standards = standardMapper.selectList();
            if (CollectionUtils.isEmpty(standards)) {
                // 读取resource目录下的device-config.json文件
                Resource resource = resourceLoader.getResource("classpath:errorStandard.json");
                // 使用fastjson2转换为DeviceConfig对象
                String jsonContent = IoUtil.readUtf8(resource.getInputStream());
                globalMemory.standard = JSON.parseArray(jsonContent, ErrorIdentificationStandard.class);
            } else {
                globalMemory.standard = standards;
            }
        }
    }

    public void pushErrorMessage(String deviceId, String dto) {
        NettyDevice device = sessionManager.getDevice(deviceId);
        if (device != null) {
            ByteBuf buf = Unpooled.buffer();
            CommandUtils.buildCommand(buf, dto, CommandConstant.interaction);
            if (device.getChannel() != null && device.getChannel().isActive()) {
                device.getChannel().writeAndFlush(buf);
            }
        }
    }


    public List<DeviceErrorPushDetail> compareErrorStandard(DeviceRunningStatusHistory history, List<ErrorIdentificationStandard> standards) {
        if (history == null || CollectionUtils.isEmpty(standards)) return null;
        List<DeviceErrorPushDetail> list = new ArrayList<>();
        standards.forEach(standard -> {
            String fieldName = standard.getField();
            String compareValues = standard.getCompareValues();
            CompareEnum compareType = standard.getCompareType();
            String targetValue = knownClassFieldFinding(fieldName, history);
            if (!StringUtils.isEmpty(compareValues) && !StringUtils.isEmpty(targetValue)) {
                boolean result = CompareUtils.compare(compareType, compareValues, targetValue);
                if (!result) {
                    DeviceErrorPushDetail detail = new DeviceErrorPushDetail();
                    detail.setErrorName(standard.getErrorName());
                    detail.setDescription(standard.getErrorDescription());
                    detail.setValue(targetValue);
                    detail.setRange(compareValues);
                    list.add(detail);
                }
            }
        });
        return list;
    }


    private static String knownClassFieldFinding(String fieldName, DeviceRunningStatusHistory history) {
        try {
            // 1. 获取类的Class对象
            Class<?> historyClass = DeviceRunningStatusHistory.class;
            Field field = historyClass.getDeclaredField(fieldName);
            field.setAccessible(true);
            Object value = field.get(history);
            return (String) value;
        } catch (NoSuchFieldException | IllegalAccessException e) {
            System.out.println("获取字段失败: " + e.getMessage());
            log.error("获取属性值失败:{}", e.getMessage());
        }
        return null;
    }

}