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

package com.maibu.service;

import java.io.IOException;
import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.maibu.constant.Constant;
import com.maibu.constant.NettyCacheKey;
import com.maibu.core.business.Device;
import com.maibu.core.business.DeviceRunningStatusHistory;
import com.maibu.core.business.DeviceStatusRecordDTO;
import com.maibu.core.business.ErrorIdentificationStandard;
import com.maibu.core.business.alarm_center.AlarmMessage;
import com.maibu.core.business.device.NettyDevice;
import com.maibu.core.business.inter.WebsocketMesDispather;
import com.maibu.core.enums.CompareEnum;
import com.maibu.core.enums.ConnectorStatus;
import com.maibu.core.enums.DeviceStatus;
import com.maibu.core.enums.ErrorSource;
import com.maibu.core.enums.RespondCode;
import com.maibu.core.host.HeartBeatDTO;
import com.maibu.core.host.LocationMessage;
import com.maibu.core.redis.RedisCache;
import com.maibu.dto.DeviceErrorPushDTO;
import com.maibu.mapper.ErrorIdentificationStandardMapper;
import com.maibu.mapper.HostLocationMapper;
import com.maibu.memory.DeviceSessionManager;
import com.maibu.memory.GlobalMemory;
import com.maibu.memory.SiteMemory;
import com.maibu.mqtt.MqttTopic;
import com.maibu.utils.CompareUtils;

import cn.hutool.core.io.IORuntimeException;
import cn.hutool.core.io.IoUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class DeviceThreadService {

    @Autowired
    private ResourceLoader resourceLoader;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private DeviceSessionManager sessionManager;

    @Autowired
    private ErrorIdentificationStandardMapper standardMapper;

    @Autowired
    private WebsocketMesDispather websocketMesDispather;

    @Autowired
    private AlertPushService alertPushService;

    @Autowired
    private IDeviceService deviceService;

    @Value("${mqtt.client-id}")
    private String mqttClientId;

    @Autowired
    private ScheduledExecutorService executor;

    @Autowired
    private HostLocationMapper hostLocationMapper;

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

        // TODO fix 后续是否分多个线程 一个site一个线程
        // 有还需要更新接受故障方式。
        executor.scheduleWithFixedDelay(() -> {
            try {
                String key = NettyCacheKey.deviceRunningStatusKey;
                Collection<String> keys = redisCache.getListKeyByPrefix(key);
                if (!CollectionUtils.isEmpty(keys)) {
                    List<DeviceErrorPushDTO> pushList = new ArrayList<>();
                    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);

                            String[] parts = k.split(":");
                            String deviceId = parts[1];
                            NettyDevice slaveDevice = sessionManager.getDevice(deviceId);
                            if (slaveDevice != null && slaveDevice.getOnlineStatus() == 1) {
                                Long siteId = slaveDevice.getDevice().getSiteId();
                                Long orgId = slaveDevice.getDevice().getOrgId();
                                log.debug("compare siteId:{},orgId:{}", siteId, orgId);
                                SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
                                if (siteMemory != null) {
                                    List<AlarmMessage> compared = compareErrorStandard(siteId, orgId, history,
                                            siteMemory.standards.get(ErrorSource.MOWER));
                                    DeviceErrorPushDTO pushDTO = new DeviceErrorPushDTO();
                                    pushDTO.setDeviceId(deviceId);
                                    pushDTO.setTime(System.currentTimeMillis());
                                    pushDTO.setDetails(compared);
                                    pushDTO.setEvent(RespondCode.device_error_push);
                                    pushList.add(pushDTO);
                                }
                            }
                        }
                    });
                    // 按实际deviceId分组，推送错误信息
                    Map<String, List<DeviceErrorPushDTO>> pushMap = pushList.stream()
                            .collect(java.util.stream.Collectors.groupingBy(DeviceErrorPushDTO::getDeviceId));
                    pushMap.forEach((deviceId, dtos) -> {
                        // pushErrorMessage(deviceId, JSON.toJSONString(dtos));
                        // todo 判断错误级别，是否需要推送邮件 或者钉钉。
                        // 找到这个设备的场站 添加场站是否开启关闭 钉钉或者邮件推送，推送邮件 添加人员是否订阅邮件，钉钉是加入钉钉群
                        alertPushService.messagePushHandler(deviceId, dtos);
                    });
                }
            } catch (Exception e) {
                log.error("执行监测错误线程错误：{}", e.getMessage());
            }
        }, 0, 5, TimeUnit.SECONDS);
    }

    public void initStandard() throws IOException {
        GlobalMemory.getAllSiteMemory().forEach(siteMemory -> {
            if (siteMemory != null && siteMemory.standards == null) {
                LambdaQueryWrapper<ErrorIdentificationStandard> queryWrapper = new LambdaQueryWrapper<>();
                queryWrapper.eq(ErrorIdentificationStandard::getSiteId, siteMemory.sysSite.getId());
                List<ErrorIdentificationStandard> list = standardMapper.selectList(queryWrapper);
                if (CollectionUtils.isEmpty(list)) {
                    try {
                        // 读取resource目录下的device-config.json文件
                        Resource resource = resourceLoader.getResource("classpath:errorStandard.json");
                        // 使用fastjson2转换为DeviceConfig对象
                        String jsonContent;
                        jsonContent = IoUtil.readUtf8(resource.getInputStream());
                        List<ErrorIdentificationStandard> newStandards = JSON.parseArray(jsonContent,
                                ErrorIdentificationStandard.class);
                        if (!CollectionUtils.isEmpty(newStandards)) {
                            newStandards.forEach(x -> {
                                x.setOrgId(siteMemory.sysSite.getOrgId());
                                x.setSiteId(siteMemory.sysSite.getId());
                            });
                            standardMapper.insertBatch(newStandards);
                            siteMemory.standards = newStandards.stream()
                                    .filter(e -> e.getErrorSource() != null)
                                    .collect(Collectors.groupingBy(
                                            ErrorIdentificationStandard::getErrorSource,
                                            ConcurrentHashMap::new,
                                            Collectors.toList()));

                        }
                    } catch (IORuntimeException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                } else {
                    siteMemory.standards = list.stream()
                            .filter(e -> e.getErrorSource() != null)
                            .collect(Collectors.groupingBy(
                                    ErrorIdentificationStandard::getErrorSource,
                                    ConcurrentHashMap::new,
                                    Collectors.toList()));
                }
            }
        });
    }

    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);
        // }
        // }
        try {
            GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId,
                    String.format(MqttTopic.DEVICE_ERROR_PUSH_TOPIC, deviceId), dto);
            // GlobalMemory.mqttClientUtil.publish(String.format(MqttTopic.DEVICE_ERROR_PUSH_TOPIC,
            // deviceId), dto);
            List<NettyDevice> controlMasters = sessionManager.getAllSlaveControl(deviceId);
            if (!CollectionUtils.isEmpty(controlMasters)) {
                controlMasters.forEach(x -> {
                    websocketMesDispather.dispather(x.getConnectorId(), dto);
                    log.info("推送错误信息 device:{},content:{}", deviceId, dto);
                });
            }
        } catch (Exception e) {
            log.error("推送错误信息失败 device:{}error:{}", deviceId, e.getMessage());
        }
    }

    public List<AlarmMessage> compareErrorStandard(Long siteId, Long orgId, DeviceRunningStatusHistory history,
            List<ErrorIdentificationStandard> standards) {
        if (history == null || CollectionUtils.isEmpty(standards))
            return null;
        List<AlarmMessage> list = new ArrayList<>();
        standards.forEach(standard -> {
            String fieldName = standard.getField();
            String compareValues = standard.getCompareValues();
            CompareEnum compareType = standard.getCompareType();
            String targetValue = knownClassFieldFinding(fieldName, history);
            log.debug("compare2  compareType:{} compareValues:{},targetValue:{}", compareType, compareValues,
                    targetValue);
            if (!StringUtils.isEmpty(compareValues) && !StringUtils.isEmpty(targetValue)) {
                boolean result = CompareUtils.compare(compareType, targetValue, compareValues);
                log.debug("compare3  result:{}", result);
                if (result) {
                    AlarmMessage alarm = new AlarmMessage();
                    alarm.setContent(standard.getErrorDescription());
                    alarm.setDeviceSn(history.getDeviceId());
                    alarm.setDeviceName(history.getDeviceId());
                    alarm.setLevel(standard.getErrorLevel());
                    alarm.setLocation(history.getLatitude() + "," + history.getLongitude());
                    alarm.setName(standard.getErrorName());
                    alarm.setNo(standard.getErrorCode());
                    alarm.setSuggestion(standard.getSuggestion());
                    alarm.setStatus("未处理");
                    alarm.setTimeValues(targetValue);
                    alarm.setTime(LocalDateTime.now());
                    alarm.setOrgId(orgId);
                    alarm.setSiteId(siteId);
                    list.add(alarm);
                }
            }
        });
        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;
    }

    /**
     * 监控心跳
     */
    public void heartBeatMonitor() {
        executor.scheduleWithFixedDelay(() -> {
            try {
                Long nowTime = System.currentTimeMillis();
                List<HeartBeatDTO> heartBeatDTOS = GlobalMemory.customMqttDeviceMonitor.getAllHeartBeat();
                if (!CollectionUtils.isEmpty(heartBeatDTOS)) {
                    heartBeatDTOS.forEach(x -> {
                        String deviceId = x.getSn();
                        Long lastActiveTime = x.getSysTimestamp();
                        NettyDevice nettyDevice = sessionManager.getDevice(x.getSn());
                        if (nowTime - lastActiveTime >= Constant.hostOfflineInterval) {
                            // 超时了判断离线
                            if (nettyDevice != null) {
                                nettyDevice.status(DeviceStatus.offline.getType());
                                // todo 推送设备离线
                                String key = NettyCacheKey.latestPositonKey + deviceId;
                                // 保存一下最新一次的数据到 数据库
                                LocationMessage locationMessage = redisCache.getCacheObject(key);
                                if (locationMessage != null) {
                                    LambdaQueryWrapper<LocationMessage> queryWrapper = new LambdaQueryWrapper<>();
                                    queryWrapper.eq(LocationMessage::getDeviceId, deviceId);
                                    LocationMessage existing = hostLocationMapper.selectOne(queryWrapper);
                                    if (existing != null) {
                                        locationMessage.setId(existing.getId());
                                    }
                                    hostLocationMapper.saveOrUpdate(locationMessage);
                                }
                            }
                        } else {
                            if (nettyDevice == null) {
                                sessionManager.registerSlaveDevice(deviceId, ConnectorStatus.ACTIVE, null);
                                Device device = deviceService.selectDeviceBySerialNumber(deviceId);
                                if (device == null) {
                                    device = new Device();
                                    device.setStatus(3);
                                    device.setOnlineStatus(DeviceStatus.online.getType());
                                    device.setSerialNumber(deviceId);
                                    device.setDeviceName(deviceId);
                                    device.setProductId(-1L);
                                    device.setTenantId(-1L);
                                    device.setFirmwareVersion(BigDecimal.valueOf(1.0));
                                    device.setProductName("割草机产品MC700");
                                    device.setCreateTime(LocalDateTime.now());
                                    device.setUpdateTime(LocalDateTime.now());
                                    device.setHasHost(true);
                                    deviceService.insertDeviceBySelf(device);
                                } else {
                                    device.setStatus(3);
                                    device.setOnlineStatus(DeviceStatus.online.getType());
                                    device.setUpdateTime(LocalDateTime.now());
                                    if (!device.isHasHost()) {
                                        device.setHasHost(true);
                                    }
                                    deviceService.updateDeviceBySelf(device);
                                }
                                // 推送json 找到设备绑定的主机进行推送
                                // List<NettyDevice> controlMasters =
                                // sessionManager.getAllSlaveControl(device.getSerialNumber());
                                // if (!CollectionUtils.isEmpty(controlMasters)) {
                                // controlMasters.forEach(x -> {
                                // DeviceStatusChangeDTO changeDTO = new DeviceStatusChangeDTO();
                                // changeDTO.setDeviceId(data);
                                // changeDTO.setStatus(DeviceStatus.online.getCode());
                                // changeDTO.setEvent(RespondCode.device_status_changed);
                                // ByteBuf buf = ctx.alloc().buffer();
                                // CommandUtils.buildCommand(buf, changeDTO, CommandConstant.interaction);
                                // String content = buf.toString(CharsetUtil.UTF_8);
                                // logger.info("推送下位机登录状态消息:{}", content);
                                // if (x.getChannel() != null && x.getChannel().isActive()) {
                                // x.getChannel().writeAndFlush(buf);
                                // }
                                // });
                            } else {
                                // todo 推送设备上线
                                nettyDevice.status(1);
                            }
                        }
                    });
                }
            } catch (Exception e) {
                log.error("执行监测心跳线程错误：{}", e.getMessage());
            }
        }, 0, 500, TimeUnit.MILLISECONDS);
    }

}