Broker/gather-broker/src/main/java/com/idtgz/config/InitConfig.java

package com.idtgz.config;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.idtgz.context.ConfigContext;
import com.idtgz.dao.entity.ClientEntity;
import com.idtgz.dao.entity.ConfigEntity;
import com.idtgz.dao.entity.ConnectionEntity;
import com.idtgz.dao.entity.ConnectionItemEntity;
import com.idtgz.dao.mapper.ClientMapper;
import com.idtgz.dao.mapper.ConfigMapper;
import com.idtgz.dao.mapper.ConnectionMapper;
import com.idtgz.mqtt.MqttServer;
import com.idtgz.service.CenterService;
import com.idtgz.service.ConnectionItemService;
import com.idtgz.service.ConnectionService;
import com.idtgz.util.CommonUtil;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import sun.nio.ch.Net;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @Description:
 * @author: GS-HQY
 * @date: 2022.05.25
 */
@Component
public class InitConfig {

    @Autowired
    ConfigMapper configMapper;

    @Autowired
    ClientMapper clientMapper;

    @Autowired
    ConnectionMapper connectionMapper;

    @Autowired
    CenterService centerService;

    @Autowired
    MqttServer scoketServer;

    @Autowired
    Scheduler scheduler;

    @Autowired
    ConnectionService connectionService;

    @Autowired
    ConnectionItemService connectionItemService;

    @PostConstruct
    public void init() throws SchedulerException {
        //配置通道
        ConfigEntity configEntity = configMapper.selectOne(new QueryWrapper<ConfigEntity>().eq("config_type", "PIPE_LINE_CONFIG"));
        if (configEntity != null && StringUtils.isNotEmpty(configEntity.getConfigJson())) {
            ConfigContext.pipeLineConfig = JSON.parseObject(configEntity.getConfigJson(), PipeLineConfig.class);
            ConfigContext.isSeted = true;
            centerService.init();
            centerService.remoteSyn();
        }

        //存储配置
        ConfigEntity storeConfigEntity = configMapper.selectOne(new QueryWrapper<ConfigEntity>().eq("config_type", "STORE_CONFIG"));
        if (storeConfigEntity != null && StringUtils.isNotEmpty(storeConfigEntity.getConfigJson())) {
            ConfigContext.storeConfig = JSON.parseObject(storeConfigEntity.getConfigJson(), StoreConfig.class);
        } else {
            StoreConfig storeConfig = new StoreConfig(3);
            ConfigContext.storeConfig = storeConfig;
            storeConfigEntity.setConfigJson(JSON.toJSONString(storeConfig));
            configMapper.updateById(storeConfigEntity);
        }

        //本地配置
        ConfigEntity localConfigEntity = configMapper.selectOne(new QueryWrapper<ConfigEntity>().eq("config_type", "LOCAL_CONFIG"));
        if (localConfigEntity != null && StringUtils.isNotEmpty(localConfigEntity.getConfigJson())) {
            ConfigContext.localConfig = JSON.parseObject(localConfigEntity.getConfigJson(), LocalConfig.class);
        } else {
            LocalConfig nLocalConfig = new LocalConfig();
            nLocalConfig.setBrokerId(UUID.randomUUID().toString().replace("-", "").toUpperCase());
            localConfigEntity.setConfigJson(JSON.toJSONString(nLocalConfig));
            configMapper.updateById(localConfigEntity);
            ConfigContext.localConfig = nLocalConfig;
        }


        //初始化客户端推送 连接
        connectionMapper.update(null, new UpdateWrapper<ConnectionEntity>().set("connection_status", ConfigConstant.CONNECTION_STATUS_OFF).eq("connection_status", ConfigConstant.CLIENT_STATUS_ON));
        List<ClientEntity> clientEntities = clientMapper.selectList(null);
        clientEntities.forEach(
                i -> {
                    i.setClientStatus(ConfigConstant.CLIENT_STATUS_OFF);
                    ConfigContext.clients.put(i.getClientId(), i);
                    clientMapper.updateById(i);
                }
        );


        //初始化连接
        List<ConnectionEntity> connectionEntities = connectionMapper.selectList(null);
        connectionEntities.forEach(
                i -> {
                    i.setConnectionStatus(ConfigConstant.CONNECTION_STATUS_OFF);
                    ConfigContext.connections.put(i.getDeviceId(), i);
                    //Item上下文
                    List<ConnectionItemEntity> itemEntities = connectionItemService.list(new QueryWrapper<ConnectionItemEntity>().eq("connection_id", i.getId()));
                    Map<String, ConnectionItemEntity> itemEntityMap = new ConcurrentHashMap<>();
                    itemEntities.forEach(
                            n -> {
                                itemEntityMap.put(n.getItemCode(), n);
                            }
                    );
                    ConfigContext.items.put(i.getDeviceId(), itemEntityMap);
                }
        );
        //quartz初始化
        quartzInit();
    }


    private void quartzInit() throws SchedulerException {
        ConfigContext.connections.forEach(
                (k, v) -> {
                    try {
                        if (v.getIsUploaded() == UploadStatus.UPLOADING && v.getConnectionStatus().equals(ConfigConstant.CONNECTION_STATUS_ON)) {
                            //非中断且连接正常则继续上传
                            connectionService.updateSchedule(v.getConnectionSign(), v.getUploadType(), v.getUploadCycle(), v.getStoreStrategy(), v.getIsUploaded(), false);
                        }
                    } catch (SchedulerException e) {
                        e.printStackTrace();
                    }
                }
        );
        scheduler.start();
    }

}