MiddlePlatform/maibu-netty-server/src/main/java/com/maibu/init/InitThread.java

package com.maibu.init;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executor;

import javax.annotation.Resource;

import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import com.maibu.constant.FastBeeConstant;
import com.maibu.core.business.DevicePlan;
import com.maibu.core.business.DevicePlanTask;
import com.maibu.core.business.DeviceRunParam;
import com.maibu.core.domain.entity.SysSite;
import com.maibu.core.enums.DeviceTaskStaus;
import com.maibu.influxdb.util.InfluxDBUtil;
import com.maibu.mapper.DevicePlanMapper;
import com.maibu.mapper.DevicePlanTaskMapper;
import com.maibu.mapper.DeviceRunParamMapper;
import com.maibu.mapper.SysSiteMapper;
import com.maibu.memory.GlobalMemory;
import com.maibu.memory.SiteMemory;
import com.maibu.mqtt.MqttClientUtil;
import com.maibu.mybatis.LambdaQueryWrapperX;
import com.maibu.service.DevicePlanTaskMonitorService;
import com.maibu.service.DeviceThreadService;
import org.springframework.beans.factory.annotation.Value;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
@Order(20)
public class InitThread implements ApplicationRunner {

    @Resource(name = FastBeeConstant.TASK.DEVICE_ERROR_MONITOR)
    private Executor deviceErrorMonitorExecutor;

    @Resource(name = FastBeeConstant.TASK.DEVICE_TASK_HANDLER)
    private Executor deviceTaskHandlerExecutor;

    @Autowired
    private DeviceThreadService deviceThreadService;

    @Autowired
    private DevicePlanTaskMonitorService devicePlanTaskMonitorService;

    @Autowired
    private DevicePlanMapper devicePlanMapper;

    @Autowired
    private DevicePlanTaskMapper devicePlanTaskMapper;

    @Autowired
    private SysSiteMapper sysSiteMapper;

    @Autowired
    private DeviceRunParamMapper deviceRunParamMapper;

    @Value("${influxdb.url}")
    private String influxDBUrl;

    @Value("${influxdb.username}")
    private String influxDBUsername;

    @Value("${influxdb.password}")
    private String influxDBPassword;

    @Value("${influxdb.bucket}")
    private String influxDBBucket;

    @Value("${influxdb.org}")
    private String influxDBOrg;

    @Value("${influxdb.token}")
    private String influxDBToken;

    @Value("${mqtt.username}")
    private String mqttUsername;

    @Value("${mqtt.password}")
    private String mqttPassword;

    @Value("${mqtt.host-url}")
    private String mqttHostUrl;

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

    @Override
    public void run(ApplicationArguments args) throws MqttException {
        recoverMemory();
        init();
    }

    private void init() throws MqttException {
        // 初始化InfluxDBUtil
        GlobalMemory.influxDBUtil = new InfluxDBUtil(influxDBUrl, influxDBToken, influxDBOrg, influxDBBucket);
        // 初始化MqttClientUtil
        GlobalMemory.mqttClientUtil = new MqttClientUtil(mqttHostUrl, mqttClientId, mqttUsername, mqttPassword);
        // 设备错误监听线程
        deviceErrorMonitorExecutor.execute(() -> {
            try {
                deviceThreadService.deviceErrorMonitor();
            } catch (InterruptedException | IOException e) {
                throw new RuntimeException(e);
            }
        });
        // 设备任务处理线程
        deviceTaskHandlerExecutor.execute(() -> {
            // 任务下发
            try {
                devicePlanTaskMonitorService.generatePlanTask();
                // 执行准备队列
                devicePlanTaskMonitorService.doLoop();
                // 执行任务
                devicePlanTaskMonitorService.doExecute();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 恢复内存
     */
    public void recoverMemory() {
        List<SysSite> siteList = sysSiteMapper.selectList();
        if (!CollectionUtils.isEmpty(siteList)) {
            siteList.forEach(x -> {
                Long orgId = x.getOrgId();
                Long siteId = x.getId();
                SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
                if (siteMemory == null) {
                    siteMemory = new SiteMemory();
                    siteMemory.setSiteId(siteId);
                    siteMemory.setSysSite(x);
                    GlobalMemory.addSiteMemory(orgId, siteId, siteMemory);
                }
            });
        }
        List<DeviceRunParam> deviceRunParamList = deviceRunParamMapper.selectList();
        if (!CollectionUtils.isEmpty(deviceRunParamList)) {
            deviceRunParamList.forEach(x -> {
                Long orgId = x.getOrgId();
                Long siteId = x.getId();
                SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
                if (siteMemory != null) {
                    siteMemory.saveDeviceRunParam(x);
                }
            });
        }
        // 未执行完的重复计划
        LambdaQueryWrapperX<DevicePlan> query = new LambdaQueryWrapperX<>();
        List<String> list = Arrays.asList(DeviceTaskStaus.NEW.getCode(), DeviceTaskStaus.EXECUTING.getCode(),
                DeviceTaskStaus.PAUSE.getCode());
        query.in(DevicePlan::getTaskStaus, list);
        List<DevicePlan> devicePlanList = devicePlanMapper.selectList(query);
        if (!CollectionUtils.isEmpty(devicePlanList)) {
            devicePlanList.forEach(x -> {
                Long orgId = x.getOrgId();
                Long siteId = x.getSiteId();
                SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
                if (siteMemory != null) {
                    siteMemory.addDevicePlan(x);
                }
            });
        }

        // 未执行完的重复计划
        LambdaQueryWrapperX<DevicePlanTask> taskQuery = new LambdaQueryWrapperX<>();
        taskQuery.eq(DevicePlanTask::getTaskStaus, DeviceTaskStaus.NEW.getCode());
        List<DevicePlanTask> newStatusList = devicePlanTaskMapper.selectList(taskQuery);
        if (!CollectionUtils.isEmpty(newStatusList)) {
            newStatusList.forEach(x -> {
                Long orgId = x.getOrgId();
                Long siteId = x.getSiteId();
                SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
                if (siteMemory != null) {
                    siteMemory.addDevicePlanPrepare(x);
                }
            });
        }

        LambdaQueryWrapperX<DevicePlanTask> executeTaskQuery = new LambdaQueryWrapperX<>();
        executeTaskQuery.eq(DevicePlanTask::getTaskStaus, DeviceTaskStaus.EXECUTING.getCode());
        List<DevicePlanTask> executeStatusList = devicePlanTaskMapper.selectList(executeTaskQuery);
        if (!CollectionUtils.isEmpty(executeStatusList)) {
            executeStatusList.forEach(x -> {
                Long orgId = x.getOrgId();
                Long siteId = x.getSiteId();
                SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
                if (siteMemory != null) {
                    siteMemory.addDevicePlanExecute(x);
                }
            });
        }

    }
}