package cc.smtweb.system.canal.example;

import cc.smtweb.framework.core.common.CanalVO;
import cc.smtweb.framework.core.common.SwEnum;
import cc.smtweb.framework.core.util.CommUtil;
import cc.smtweb.framework.core.util.DateUtil;
import cc.smtweb.framework.core.util.JsonUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.io.File;
import java.io.FileInputStream;
import java.net.InetSocketAddress;
import java.util.*;

/**
 * @Author yaoq
 * @Date 2022年09月13日 18:02
 * @Description
 */
public class ClientInstance {
    private static final String CLASSPATH_URL_PREFIX = "classpath:";
    private static final Logger logger = LoggerFactory.getLogger(ClientInstance.class);
    protected volatile boolean running = false;
    protected Thread thread = null;
    protected CanalConnector connector;
    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {

        public void uncaughtException(Thread t, Throwable e) {
            logger.error("解析出错了哦", e);

        }
    };

    private static ClientInstance instance = null;
    private static Properties properties = null;

    static {
        instance = new ClientInstance();
    }

    public static ClientInstance getInstance() {
        return instance;
    }

    private void loadProperties() {
        try {
            String conf = System.getProperty("client.conf", "classpath:client.properties");
            properties = new Properties();
            if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
                conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
                properties.load(ClientInstance.class.getClassLoader().getResourceAsStream(conf));
            } else {
                properties.load(new FileInputStream(conf));
            }
        } catch (Exception e) {
            logger.error("加载canal客户端配置参数失败！", e);
        }
    }

    private ClientInstance() {
        loadProperties();
        connector = CanalConnectors.newSingleConnector(new
                InetSocketAddress(ClientConsts.getIp(properties),
                ClientConsts.getPort(properties)), ClientConsts.getInstance(properties), ClientConsts.getUsername(properties), ClientConsts.getPassword(properties));
        //connector.connect();
        //connector.subscribe(ClientConsts.getFilter(properties));
        //connector.rollback();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                logger.info("## stop the canal client");
                ClientInstance.this.stop();
            } catch (Throwable e) {
                logger.warn("##something goes wrong when stopping canal:", e);
            } finally {
                logger.info("## canal client is down.");
            }
        }));
    }

    //开始
    public void start() {
        Assert.notNull(connector, "connector is null");
        thread = new Thread(() -> process());

        thread.setUncaughtExceptionHandler(handler);
        running = true;
        thread.start();
    }

    //停止
    public void stop() {
        if (!running) {
            return;
        }
        running = false;
        if (thread != null) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                // ignore
            }
        }
    }

    protected void process() {
        final int batchSize = ClientConsts.getBatchSize(properties);
        while (running) {
            try {
                connector.disconnect();
                connector.connect();
                connector.unsubscribe();
                connector.subscribe(ClientConsts.getFilter(properties));
                while (running) {
                    Message message = connector.getWithoutAck(batchSize * 1024); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        Thread.sleep(10 * 1000L);//无数据时，睡眠10秒。
                        continue;
                    } else {
                        dataHandle(message.getEntries());
                    }
                    if (batchId != -1) {
                        connector.ack(batchId); // 提交确认
                    }
                }
            } catch (Throwable e) {
                logger.error("process error!", e);
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e1) {
                    // ignore
                }
                if (!isReConnection()) {
                    connector.rollback(); // 处理失败, 回滚数据
                }
            } finally {
                connector.disconnect();
            }
        }
    }

    private boolean isReConnection() {
        if (!connector.checkValid()) return false;
        int count = 1;
        int rate = 1;
        while (connector.checkValid()) {
            rate = (count + 100) / 100;
            try {
                logger.debug("canal服务连接中断，第【" + count + "】次重连开始");
                connector.disconnect();
                connector.connect();
                logger.debug("canal服务连接中断，第【" + count + "】次重连成功");
                break;
            } catch (Throwable e2) {
                if (count > 99) {
                    logger.error("canal服务连接异常！", "重新连接canal服务端已失败【" + count + "】次，请检查canal服务是否启动，请及时启动canal服务端，或联系研发人员！");
                }
                logger.debug("canal服务连接中断，第【" + count + "】次重连失败：" + CommUtil.getOrigMsg(e2));
                logger.debug("canal服务连接中断，" + (1 * rate) + "分钟后进行第【" + (count + 1) + "】连接测试！");
                try {
                    Thread.sleep(60 * rate * 1000L);
                } catch (InterruptedException e1) {
                    // ignore
                }
            }
            count++;
        }
        return true;
    }

    /**
     * 数据处理
     *
     * @param entrys
     */
    private void dataHandle(List<CanalEntry.Entry> entrys) throws Exception {
        List<CanalVO> data = new ArrayList<>();
        for (CanalEntry.Entry entry : entrys) {
            if (CanalEntry.EntryType.ROWDATA != entry.getEntryType()) continue;
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            CanalEntry.EventType eventType = rowChange.getEventType();
            CanalVO row = null;
            if (eventType == CanalEntry.EventType.CREATE) {
                row = saveCreateSql(entry, rowChange);
            } else if (eventType == CanalEntry.EventType.ALTER) {
                row = saveAlterSql(entry, rowChange);
            } else if (eventType == CanalEntry.EventType.DELETE) {
                row = saveDeleteSql(entry, rowChange);
            } else if (eventType == CanalEntry.EventType.UPDATE) {
                row = saveUpdateSql(entry, rowChange);
            } else if (eventType == CanalEntry.EventType.INSERT) {
                row = saveInsertSql(entry, rowChange);
            }
            if (row == null) continue;
            data.add(row);
        }
        CanalEntry.Entry entry = entrys.get(entrys.size() - 1);
        logger.debug("sync data size：" + data.size() + "，canal position：filename：" + entry.getHeader().getLogfileName() + ",offset：" + entry.getHeader().getLogfileOffset());
        if (data.size() == 0) return;
        saveToFile(data);
    }


    /**
     * 保存数据表新增语句
     *
     * @param entry
     */
    private CanalVO saveCreateSql(CanalEntry.Entry entry, CanalEntry.RowChange rowChange) {
        try {
            return CanalVO.ok(entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), SwEnum.DbEventType.CREATE.value, null, rowChange.getSql(), null, null);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     * 保存结构变更语句
     *
     * @param entry
     */
    private CanalVO saveAlterSql(CanalEntry.Entry entry, CanalEntry.RowChange rowChange) {
        try {
            return CanalVO.ok(entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), SwEnum.DbEventType.ALTER.value, null, rowChange.getSql(), null, null);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 保存插入语句
     *
     * @param entry
     */
    private CanalVO saveInsertSql(CanalEntry.Entry entry, CanalEntry.RowChange rowChange) {
        try {
            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDatasList) {
                List<CanalEntry.Column> columnList = rowData.getAfterColumnsList();
                Map<String, String> data = new HashMap<>();
                columnList.stream().forEach(item -> {
                    data.put(item.getName(), item.getValue());
                });
                return CanalVO.ok(entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), SwEnum.DbEventType.INSERT.value, data, null, null, null);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 保存更新语句
     *
     * @param entry
     */
    private CanalVO saveUpdateSql(CanalEntry.Entry entry, CanalEntry.RowChange rowChange) {
        try {
            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDatasList) {
                List<CanalEntry.Column> newColumnList = rowData.getAfterColumnsList();
                Map<String, String> data = new HashMap<>();
                newColumnList.stream().forEach(item -> {
                    data.put(item.getName(), item.getValue());
                });
                List<CanalEntry.Column> oldColumnList = rowData.getBeforeColumnsList();
                return CanalVO.ok(entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), SwEnum.DbEventType.UPDATE.value, data, null, oldColumnList.get(0).getName(), oldColumnList.get(0).getValue());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 保存删除语句
     *
     * @param entry
     */
    private CanalVO saveDeleteSql(CanalEntry.Entry entry, CanalEntry.RowChange rowChange) {
        try {
            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDatasList) {
                List<CanalEntry.Column> oldColumnList = rowData.getBeforeColumnsList();
                return CanalVO.ok(entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), SwEnum.DbEventType.DELETE.value, null, null, oldColumnList.get(0).getName(), oldColumnList.get(0).getValue());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    private static synchronized File createFile(String basePath) throws Exception {
        String path = basePath;
        if (!path.endsWith("/")) {
            path = path + "/";
        }
        path = path + "increment/";
        File file = new File(path + "/" + DateUtil.toLongDateTimeString(null) + ".cf");
        if (!file.getParentFile().exists()) { // 如果父目录不存在，创建父目录
            file.getParentFile().mkdirs();
        }
        while (file.exists()) {
            Thread.sleep(1);
            file = new File(path + "/" + DateUtil.toLongDateTimeString(null) + ".cf");
            /*if (file.exists()) {
                return file;
            }*/
        }
        file.createNewFile();
        if (file.exists()) logger.debug("canal文件生成成功：" + file.getAbsolutePath());
        return file;
    }

    private void saveToFile(List<CanalVO> rows) throws Exception {
        // 将格式化后的字符串写入文件
        FileUtils.writeStringToFile(createFile(ClientConsts.getFilepath(properties)), JsonUtil.encodeString(rows), "UTF-8");
    }

}


