package com.yohoufo.common.cache;

import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.yoho.core.redis.cluster.annotation.Redis;
import com.yoho.core.redis.cluster.operations.nosync.YHHashOperations;
import com.yoho.core.redis.cluster.operations.nosync.YHRedisTemplate;
import com.yoho.core.redis.cluster.operations.nosync.YHValueOperations;
import com.yoho.core.redis.cluster.operations.serializer.RedisKeyBuilder;

public class RedisGwCacheClient implements CacheClient {
	
    private static final Logger LOGGER = LoggerFactory.getLogger(RedisGwCacheClient.class);
	
	@Redis("gwNoSyncRedis")
	private YHRedisTemplate redis;
	
	@Redis("gwNoSyncRedis")
	private YHValueOperations valueOperations;

	@Redis("gwNoSyncRedis")
	private YHHashOperations hashOperations;
	
	@Override
	public void set(String key, int expireInSeconds, Object o) {
		if (expireInSeconds > 120) {
			expireInSeconds = expireInSeconds + getRandomNum(-20, 60);
		}
		try {
			String compressStr= SnappyZipUtils.compress(JSON.toJSONString(o));
			this.valueOperations.set(RedisKeyBuilder.newInstance().appendFixed(key), compressStr,
					expireInSeconds, TimeUnit.SECONDS);
		} catch (Exception e) {
			LOGGER.warn("set into redis failed!! key is :{} ,e is :{}", key,e);
			// do nothings
		}
	}

	@Override
	public void setAndSync(String key, int expireInSeconds, Object o) {
		if (expireInSeconds > 120) {
			expireInSeconds = expireInSeconds + getRandomNum(-20, 60);
		}
		try {
			String compressStr= SnappyZipUtils.compress(JSON.toJSONString(o));
			this.valueOperations.setAndAsyncClean(RedisKeyBuilder.newInstance().appendFixed(key), compressStr,
					expireInSeconds, TimeUnit.SECONDS);
		} catch (Exception e) {
			LOGGER.warn("set into redis failed!! key is :{} ,e is :{}", key,e);
			// do nothing
		}
	}

	@Override
	public <T> T get(String key, Class<T> clazz) {
		try {
			String value = this.valueOperations.get(RedisKeyBuilder.newInstance().appendFixed(key));
			if (StringUtils.isNotEmpty(value)) {
				String unCompressStr= SnappyZipUtils.uncompress(value);
				return JSON.parseObject(unCompressStr, clazz);
			} else {
				LOGGER.debug("can not get data from redis by key {}.", key);
				return null;
			}
		} catch (Exception e) {
			LOGGER.warn("get from redis failed!! key is:{},e is :{}", key,e);
			return null;
		}
	}

	@Override
	public <T> Map<String, T> getBulk(Collection<String> keys, Class<T> clazz) {
		List<RedisKeyBuilder> redisKeyBuilders=convert(keys);
		try {
			List<String> multiGet =valueOperations.multiGet(redisKeyBuilders);
			int size = multiGet.size();
			Map<String, T> result = new HashMap<String, T>(size);
			String curItem;
			for (int i = 0; i < size; i++) {
				curItem = multiGet.get(i);
				if (null == curItem) {
					result.put(redisKeyBuilders.get(i).getKey(), null);
				} else {
					result.put(redisKeyBuilders.get(i).getKey(), JSON.parseObject(curItem, clazz));
				}
			}
			return result;
		} catch (Exception e) {
			LOGGER.warn("getBulk from redis failed!!", e);
			return Maps.newHashMap();
		} 
	}
	
	@Override
	public void delete(String key) {
		try {
			redis.delete(RedisKeyBuilder.newInstance().appendFixed(key));
		} catch (Exception e) {
			LOGGER.warn("delete from redis failed!! key is:{},e is :{}", key,e);
		}
	}

	@Override
	public void mset(Map<String, ? extends Object> map, long timeout) {
		if (MapUtils.isEmpty(map)) {
			return;
		}
		Map<RedisKeyBuilder, String> cacheMap = new HashMap<RedisKeyBuilder, String>(map.size());
		try{

			for (String key : map.keySet()) {
				Object value = map.get(key);
				RedisKeyBuilder keyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
				if (value instanceof String) {
					cacheMap.put(keyBuilder, (String) value);
				} else if (value instanceof Integer) {
					cacheMap.put(keyBuilder, value + "");
				} else {
					cacheMap.put(keyBuilder, JSON.toJSONString(value));
				}
			}
			redis.mset(cacheMap, timeout);
		}catch (Exception e){
			LOGGER.warn("mset cache failed!!! e is: {}", e);
		}
	}

	@Override
	public <T> void hashPut(String key, String field, T value, long timeout) {
		if(null == value){
			return;
		}
		RedisKeyBuilder keyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
		try{
			String hashValue;
			if (value instanceof String){
				hashValue = (String)value;
			}else{
				hashValue = JSON.toJSONString(value);
			}
			hashOperations.put(keyBuilder, field, hashValue);
			redis.longExpire(keyBuilder, timeout, TimeUnit.SECONDS);
		}catch (Exception e){
			LOGGER.warn("hashPut redis value operation failed. e is: {}", e);
		}
	}

	@Override
	public <T> T hashGet(String key, String field, Class<T> clazz){
		String value = null;
		try {
			RedisKeyBuilder keyBuilder = RedisKeyBuilder.newInstance().appendFixed(key);
			value = hashOperations.get(keyBuilder, field);
			return getValue(value, clazz);
		}catch (Exception e){
			LOGGER.warn("hashGet redis value operation failed. e is: {}", e);
		}
		return null;
	}
	
	@Override
	public boolean setNx(String key, int expireInSeconds, Object o) {
		try {
			RedisKeyBuilder keyB = RedisKeyBuilder.newInstance().appendFixed(key);
			String compressStr = SnappyZipUtils.compress(JSON.toJSONString(o));
			Boolean resu = this.valueOperations.setIfAbsent(keyB, compressStr);
			if (resu != null && resu) {
				redis.expire(keyB, expireInSeconds, TimeUnit.SECONDS);
				return true;
			}
			return false;
		} catch (Exception e) {
			LOGGER.warn("setNx into redis failed!! key is :{} ,e is :{}", key, e);
			return false;
		}
	}
	

	  /**
     * 获取正负区间的一个随机数
     *
     * @param smallistNum
     * @param biggestNum
     * @return
     */
    private int getRandomNum(int smallistNum, int biggestNum) {
        Random random = new Random();
        return (Math.abs(random.nextInt()) % (biggestNum - smallistNum + 1)) + smallistNum;
    }

	private List<RedisKeyBuilder> convert(Collection<String> keys) {
		return keys.stream().map(key -> RedisKeyBuilder.newInstance().appendFixed(key)).collect(Collectors.toList());
	}

	@SuppressWarnings("unchecked")
	private <T> T getValue(String value, Class<T> clazz) {
		T t = null;
		if (value == null){
			return (T) null;
		}
		if (clazz.equals(String.class)) {
			t = (T) value;
		} else if (clazz.equals(Integer.class)) {
			t = (T) Integer.valueOf(value);
		} else {
			try{
				t = JSON.parseObject(value, clazz);
			}catch (Exception e) {
				LOGGER.warn("get value cache failed!!! value is: "+ value,e);
			}
		}
		return t;
	}
}