package com.sunwin.metro.source;

import com.sunwin.metro.utils.MysqlUtil;
import com.sunwin.metro.utils.StringUtils;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import pro.husk.mysql.MySQL;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @program: sunwin_metro
 * @description: 从mysql读取设备名，用来模拟设备数据
 * @author: xuYJ
 * @create: 2021-12-01 15:26
 **/
public class MetroSubSource extends RichSourceFunction<Map<String, List<String>>> {
    private volatile boolean isRunning = true;
    private MySQL sql;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        sql = MysqlUtil.getProMeter();
    }

    @Override
    public void close() throws Exception {
        if (null != sql) {
            sql.closeConnection();
        }
        super.close();
    }

    @Override
    public void run(SourceContext<Map<String, List<String>>> ctx) throws Exception {
        while (isRunning) {
            Map<String, List<String>> keyWords = new HashMap<>(16);
            try {
                //从Mysql数据库获取配置
                List<String> list = new ArrayList<>();
                // Execute query
                try (ResultSet results = sql.query("select tl.line_name as sub_name,  \n" +
                        "                       ts.station_name as sub_station, \n" +
                        "                         concat(te.name,'-',t.equipment_id) as sub_metro,  \n" +
                        "                       ta.id as app_code \n" +
                        "                       from t_equipment t \n" +
                        "                      LEFT JOIN t_equipment_type te on t.equipment_type = te.id \n" +
                        "                        LEFT JOIN t_station ts on t.station_id = ts.id\n" +
                        "                       LEFT JOIN t_line tl on tl.id = ts.line_id\n" +
                        "                      LEFT JOIN t_application ta on ta.id = t.app_id")) {
                    while (results.next()) {
                        String numSource = results.getString("sub_name");
                        String station = results.getString("sub_station");
                        String metro = results.getString("sub_metro");
                        String appType = results.getString("app_code");
                        list.add(numSource + "_" + station + "_" + metro+"_"+appType);
                    }
                    results.getStatement().close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                //解析规则与业务逻辑相关请忽略
                if (!StringUtils.isNullOrEmpty(list)) {
                    keyWords.put("keyWords", list);
                    //输出规则流
                    ctx.collect(keyWords);
                }
            } catch (Exception e) {
                System.out.println("连接mysql模拟设备库异常");
                e.printStackTrace();
                sql = MysqlUtil.getProMeter2();
            }
            //线程睡眠(半个小时刷新一次mysql)
            Thread.sleep(5000L);
        }
    }

    @SneakyThrows
    @Override
    public void cancel() {
        isRunning = false;
        if (null != sql) {
            sql.closeConnection();
        }
    }
}
