package com.bellszhu.elasticsearch.plugin.synonym.analysis;
import com.bellszhu.elasticsearch.plugin.DynamicSynonymPlugin;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.synonym.SynonymMap;
import org.elasticsearch.common.io.PathUtils;
import org.elasticsearch.env.Environment;
import java.io.*;
import java.nio.file.Path;
import java.sql.*;
import java.util.ArrayList;
import java.util.Properties;

public class MySqlRemoteSynonymFile implements SynonymFile{
    
    private final static String DB_PROPERTIES = "jdbc-reload.properties";
    private static Logger logger = LogManager.getLogger("dynamic-synonym");
    private String format;
    private boolean expand;
    private boolean lenient;
    private Analyzer analyzer;
    private Environment env;
    // 数据库配置
    private String location;
    // 数据库地址
    private static final String jdbcUrl = "jdbc.url";
    // 数据库用户名
    private static final String jdbcUser = "jdbc.user";
    // 数据库密码
    private static final String jdbcPassword = "jdbc.password";
    
    private long thisSynonymVersion = -1L;
    private Connection connection = null;
    private Statement statement = null;
    private Properties props;
    private Path conf_dir;
    MySqlRemoteSynonymFile(Environment env, Analyzer analyzer,
                           boolean expand, boolean lenient, String format, String location) {
        this.analyzer = analyzer;
        this.expand = expand;
        this.format = format;
        this.lenient = lenient;
        this.env = env;
        this.location = location;
        this.props = new Properties();
        //读取当前 jar 包存放的路径
        Path filePath = PathUtils.get(new File(DynamicSynonymPlugin.class.getProtectionDomain().getCodeSource()
                .getLocation().getPath())
                .getParent(), "config")
                .toAbsolutePath();
        this.conf_dir = filePath.resolve(DB_PROPERTIES);
        //判断文件是否存在
        File configFile = conf_dir.toFile();
        InputStream input = null;
        try {
            input = new FileInputStream(configFile);
        } catch (FileNotFoundException e) {
            logger.info("jdbc-reload.properties 数据库配置文件没有找到， " + e);
        }
        if (input != null) {
            try {
                props.load(input);
            } catch (IOException e) {
                logger.error("数据库配置文件 jdbc-reload.properties 加载失败，" + e);
            }
        }
        isNeedReloadSynonymMap();
    }
    
    @Override
    public SynonymMap reloadSynonymMap() {
        try {
            logger.info("start reload local synonym from {}.", location);
            Reader rulesReader = getReader();
            SynonymMap.Builder parser = RemoteSynonymFile.getSynonymParser(rulesReader, format, expand, lenient, analyzer);
            return parser.build();
        } catch (Exception e) {
            logger.error("reload local synonym {} error!", e, location);
            throw new IllegalArgumentException(
                    "could not reload local synonyms file to build synonyms", e);
        }
    }
    
    @Override
    public boolean isNeedReloadSynonymMap() {
        try {
            Long mysqlVersion = getMySqlSynonymVersion();
            if (thisSynonymVersion < mysqlVersion) {
                thisSynonymVersion = mysqlVersion;
                return true;
            }
        } catch (Exception e) {
            logger.error(e);
        }
        return false;
    }
    
    public Long getMySqlSynonymVersion() {
        ResultSet resultSet = null;
        Long mysqlSynonymVersion = 0L;
        try {
            if (connection == null || statement == null) {
//                Class.forName(props.getProperty("jdbc.driver"));
                statement = getConnection(props, connection);
            }
            resultSet = statement.executeQuery(props.getProperty("jdbc.reload.swith.synonym.version"));
            while (resultSet.next()) {
                mysqlSynonymVersion = resultSet.getLong("swith_state");
                logger.info("当前MySql同义词版本号为:{}, 当前节点同义词库版本号为:{}", mysqlSynonymVersion, thisSynonymVersion);
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            try {
                if (resultSet != null) {
                    resultSet.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return mysqlSynonymVersion;
    }
    
    public ArrayList<String> getDBData() {
        ArrayList<String> arrayList = new ArrayList<>();
        ResultSet resultSet = null;
        try {
            if (connection == null || statement == null) {
//                Class.forName(props.getProperty("jdbc.driver"));
                statement = getConnection(props, connection);
            }
            logger.info("正在执行SQL查询同义词列表，SQL:{}", props.getProperty("jdbc.reload.synonym.sql"));
            resultSet = statement.executeQuery(props.getProperty("jdbc.reload.synonym.sql"));
            while (resultSet.next()) {
                String theWord = resultSet.getString("words");
                arrayList.add(theWord);
            }
        } catch (SQLException e) {
            logger.error(e);
        } finally {
            try {
                if (resultSet != null) {
                    resultSet.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return arrayList;
    }
    
    @Override
    public Reader getReader() {
        StringBuffer sb = new StringBuffer();
        try {
            ArrayList<String> dbData = getDBData();
            for (int i = 0; i < dbData.size(); i++) {
                logger.info("正在加载同义词:{}", dbData.get(i));
                // 获取一行一行的记录，每一条记录都包含多个词，形成一个词组，词与词之间使用英文逗号分割
                sb.append(dbData.get(i))
                        .append(System.getProperty("line.separator"));
            }
        } catch (Exception e) {
            logger.error("同义词加载失败");
        }
        return new StringReader(sb.toString());
    }
    
    private static Statement getConnection(Properties props, Connection conn) throws SQLException {
        conn = DriverManager.getConnection(
                props.getProperty(jdbcUrl),
                props.getProperty(jdbcUser),
                props.getProperty(jdbcPassword));
        return conn.createStatement();
    }
}

    SynonymFile getSynonymFile(Analyzer analyzer) {
        try {
            SynonymFile synonymFile;
            if ("fromMySql".equals(location)) {
                synonymFile = new MySqlRemoteSynonymFile(environment, analyzer, expand, lenient, format, location);
            }else if (location.startsWith("http://") || location.startsWith("https://")) {
                synonymFile = new RemoteSynonymFile(
                        environment, analyzer, expand, lenient,  format, location);
            } else {
                synonymFile = new LocalSynonymFile(
                        environment, analyzer, expand, lenient, format, location);
            }
            if (scheduledFuture == null) {
                scheduledFuture = pool.scheduleAtFixedRate(new Monitor(synonymFile),
                                interval, interval, TimeUnit.SECONDS);
            }
            return synonymFile;
        } catch (Exception e) {
            logger.error("failed to get synonyms: " + location, e);
            throw new IllegalArgumentException("failed to get synonyms : " + location, e);
        }
    }

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.38</version>
</dependency>

<?xml version="1.0"?>
<assembly>
    <id>-</id>
    <formats>
        <format>zip</format>
    </formats>
    <includebaseDirectory>false</includebaseDirectory>
    <fileSets>
        <fileSet>
            <directory>${project.basedir}/config</directory>
            <outputDirectory>config</outputDirectory>
        </fileSet>
    </fileSets>
    <files>
        <file>
            <source>${project.basedir}/src/main/resources/plugin-descriptor.properties</source>
            <outputDirectory>/</outputDirectory>
            <filtered>true</filtered>
        </file>
        <file>
            <source>${project.basedir}/src/main/resources/plugin-security.policy</source>
            <outputDirectory>/</outputDirectory>
            <filtered>true</filtered>
        </file>
    </files>
...略...
</assembly>

PUT synonyms_index
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0,
    "index": {
      "analysis": {
        "filter": {
          "mysql_synonym": {
            "type": "dynamic_synonym",
            "synonyms_path": "fromMySql",
            "interval": 30
          }
        },
        "analyzer": {
          "ik_syno": {
            "type": "custom",
            "tokenizer": "ik_smart",
            "filter": [
              "mysql_synonym"
            ]
          },
          "ik_syno_max": {
            "type": "custom",
            "tokenizer": "ik_max_word",
            "filter": [
              "mysql_synonym"
            ]
          }
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text",
        "analyzer": "ik_syno_max",
        "search_analyzer": "ik_syno"
      },
      "title": {
        "type": "text",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart"
      }
    }
  }
}