package com.yiidata.intergration.common.cache;

import com.yiidata.intergration.common.cache.serde.JdkSerializer;
import com.yiidata.intergration.common.cache.serde.Serializer;
import com.yiidata.intergration.common.cache.serde.StringSerializer;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.io.Closeable;
import java.io.IOException;
import java.io.Serializable;
import java.util.Set;

/**
 * <pre>
 *
 * Created by zhaopx.
 * User: zhaopx
 * Date: 2019/8/21
 * Time: 18:55
 *
 * </pre>
 *
 * @author zhaopx
 */
public class Redised implements ICache<Serializable>, Closeable {


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



    /**
     * Redis 的 Java Client
     */
    JedisPool redisPool;

    /**
     * redis db index
     */
    private final String host;

    /**
     * redis db index
     */
    private final int port;

    /**
     * redis db password
     */
    private final String password;

    /**
     * redis db index
     */
    private final int db;

    /**
     * KEY 序列化
     */
    private final Serializer<String> KEY_SERDE = new StringSerializer();

    /**
     * Redis Cache 序列化方式
     */
    private Serializer<Serializable> VALUE_SERDE = new JdkSerializer();



    public Redised() {
        this("localhost", 6379, 0);
    }


    public Redised(String host, int port){
        this(host, port, 0);
    }


    public Redised(String host, int port, int db){
        this(host, port, null, db);
    }


    public Redised(String host, int port, String password){
        this(host, port, password, 0);
    }


    public Redised(String host, int port, String password, int db){
        this.db = db;
        this.host = host;
        this.port = port;
        this.password = password;

        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxIdle(100);
        poolConfig.setMinIdle(5);

        //jedis = new Jedis(host, port, 30000, 3000);
        if(StringUtils.isNotBlank(password)) {
            this.redisPool = new JedisPool(poolConfig, host, port, 60000, password);
        } else {
            this.redisPool = new JedisPool(poolConfig, host, port);
        }
    }

    @Override
    public void add(String key, Serializable value) {
        if(value == null) {
            return;
        }
        try(Jedis jedis = redisPool.getResource();) {
            jedis.select(db);
            jedis.set(KEY_SERDE.serialize(key), VALUE_SERDE.serialize(value));
        }
    }

    @Override
    public void add(String key, int exp, Serializable value) {
        byte[] keybytes = KEY_SERDE.serialize(key);
        try(Jedis jedis = redisPool.getResource();) {
            jedis.select(db);
            jedis.set(keybytes, VALUE_SERDE.serialize(value));
            jedis.expire(keybytes, exp);
        }
    }

    @Override
    public Serializable get(String key) {
        if(key == null) {
            return null;
        }
        try(Jedis jedis = redisPool.getResource();) {
            jedis.select(db);
            byte[] bytes = jedis.get(KEY_SERDE.serialize(key));
            if (bytes == null) {
                return null;
            }
            return VALUE_SERDE.deserialize(bytes);
        }
    }


    /**
     * 移除某条缓存
     * @param key 缓存的 KEY
     * @return
     */
    @Override
    public Serializable remove(String key) {
        if(key == null) {
            return null;
        }
        logger.info("remove cache key: {}", key);
        try(Jedis jedis = redisPool.getResource();) {
            jedis.select(db);
            return jedis.del(KEY_SERDE.serialize(key));
        }
    }


    @Override
    public int removeByPrefix(String prefix) {
        try(Jedis jedis = redisPool.getResource();) {
            jedis.select(db);
            Set<String> keys = jedis.keys(prefix + "*");
            if (keys != null && keys.size() > 0) {
                for (String key : keys) {
                    remove(key);
                }
            }
            return keys == null ? 0 : keys.size();
        }
    }

    @Override
    public void clear() {
        try(Jedis jedis = redisPool.getResource();) {
            jedis.select(db);
            jedis.flushDB();
        }
    }

    @Override
    public int size() {
        try(Jedis jedis = redisPool.getResource();) {
            jedis.select(db);
            return jedis.dbSize().intValue();
        }
    }

    @Override
    public void close() throws IOException {
        redisPool.close();
    }

    @Override
    public void shutdown() {
        try {
            close();
        } catch (IOException e) {

        }
    }

    /**
     * 序列化方式
     * @param valueSerde
     */
    public void setValueSerde(Serializer<Serializable> valueSerde) {
        this.VALUE_SERDE = valueSerde;
    }

    public String getHost() {
        return host;
    }

    public int getPort() {
        return port;
    }

    public String getPassword() {
        return password;
    }

    public int getDb() {
        return db;
    }

    public JedisPool getRedisPool() {
        return redisPool;
    }
}