package com.alibaba.otter.canal.client.adapter.clickhouse.support;

import java.io.Closeable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * sql批量执行器
 *
 * @author rewerma 2018-11-7 下午06:45:49
 * @version 1.0.0
 */
public class BatchExecutor implements Closeable {

    private static final Logger logger = LoggerFactory.getLogger(BatchExecutor.class);

    private DataSource          dataSource;
    private Connection          conn;
    private AtomicInteger       idx    = new AtomicInteger(0);

    public BatchExecutor(DataSource dataSource){
        this.dataSource = dataSource;
    }

    public Connection getConn() {
        if (conn == null) {
            try {
                conn = dataSource.getConnection();
                this.conn.setAutoCommit(false);
            } catch (SQLException e) {
                logger.error(e.getMessage(), e);
            }
        }
        return conn;
    }

    public static void setValue(List<Map<String, ?>> values, int type, Object value) {
        Map<String, Object> valueItem = new HashMap<>();
        valueItem.put("type", type);
        valueItem.put("value", value);
        values.add(valueItem);
    }

    public void execute(String sql, List<Map<String, ?>> values) throws SQLException {
        PreparedStatement pstmt = getConn().prepareStatement(sql);
        int len = values.size();
        for (int i = 0; i < len; i++) {
            int type = (Integer) values.get(i).get("type");
            Object value = values.get(i).get("value");
            SyncUtil.setPStmt(type, pstmt, value, i + 1);
        }

        pstmt.execute();
        idx.incrementAndGet();
        pstmt.close();
    }

    public void batchExecute(String sql, List<List<Map<String, ?>>> batchValues) throws SQLException {
        PreparedStatement pstmt = getConn().prepareStatement(sql);
        for (int i = 0; i < batchValues.size(); i++) {
            List<Map<String, ?>> values = batchValues.get(i);
            for (int j = 0; j < values.size(); j++) {
                int type = (Integer) values.get(j).get("type");
                Object value = values.get(j).get("value");
                SyncUtil.setPStmt(type, pstmt, value, j + 1);
            }
            pstmt.addBatch();
        }

        pstmt.executeBatch();
        idx.addAndGet(batchValues.size());
        pstmt.close();
    }

    public void commit() throws SQLException {
        getConn().commit();
        if (logger.isTraceEnabled()) {
            logger.trace("Batch executor commit " + idx.get() + " rows");
        }
        idx.set(0);
    }

    public void rollback() throws SQLException {
        getConn().rollback();
        if (logger.isTraceEnabled()) {
            logger.trace("Batch executor rollback " + idx.get() + " rows");
        }
        idx.set(0);
    }

    @Override
    public void close() {
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                logger.error(e.getMessage(), e);
            } finally {
                conn = null;
            }
        }
    }
}
