package project.redis.interal;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;

import kernel.util.ThreadUtils;
import project.redis.RedisHandler;

public class RedisHandlerImpl implements RedisHandler, InitializingBean, Runnable {
	
	private Redis redis;
	
	private TaskExecutor taskExecutor;
	
	private Logger logger = LoggerFactory.getLogger(RedisHandlerImpl.class);

	/**
	 * get
	 * @param key
	 * @param object
	 */
	public Object get(String key) {
		return redis.get(key);
	}

	/**
	 * 批量get，与单个get存在性能区别，一次连接（redispool），遍历取到数据后返回
	 */
	public Object[] getList(String[] keys) {
		return redis.getList(keys);
	}
	
	@Override
	public <V> HashMap<String,V> getMap(Set<String> keys) {
		return redis.getMap(keys);
	}
	
	/**
	 * set 同步
	 * 
	 * @param key
	 * @param object
	 */
	public void setSync(String key, Object object) {
		redis.setSync(key, object);
	}

	/**
	 * set 批量同步
	 * 
	 * @param params 需要写入的 k-v 数据
	 */
	public void setBatchSync(Map<String, Object> params) {
		redis.setBatchSync(params);
	}

	/**
	 * set 异步
	 * 
	 * @param key
	 * @param object
	 */
	public void setAsyn(String key, Object object) {
		AsynItem item = new AsynItem(key, object, AsynItem.TYPE_MAP);
		AsynItemQueue.add(item);
	}

	public void remove(String key) {
		redis.remove(key);
	}

	/**
	 * push 同步
	 */
	public void pushSync(String key, Object object) {
		redis.pushSync(key, object);
	}

	/**
	 * push 异步。批量处理在业务里考虑
	 * 
	 */
	public void pushAsyn(String key, Object object) {
		AsynItem item = new AsynItem(key, object, AsynItem.TYPE_QUEUE);
		AsynItemQueue.add(item);
	}

	/**
	 * push 异步。批量处理在业务里考虑
	 * 
	 */
	public void pushBatchAsyn(List<Map<String, Object>> params) {
		redis.pushBatchSync(params);
	}

	/**
	 * 从队列尾取一个Object
	 * 
	 */
	public Object poll(String key) {
		return redis.poll(key);
	}

	/**
	 * 服务运行： 1. 从消息队列获取message 2.调用currentProvider发送短信
	 */
	public void run() {
		List<AsynItem> list = new ArrayList<AsynItem>();
		while (true) {

			try {
				
				AsynItem item = AsynItemQueue.poll();
				if (item != null) {
					list.add(item);
				}

				if ((item == null && list.size() > 0) || list.size() >= 100) {
					taskExecutor.execute(new HandleRunner(list));
					list = new ArrayList<AsynItem>();
				}
				if (item == null) {
					ThreadUtils.sleep(50);
				}
			} catch (Throwable e) {
				logger.error("RedisHandlerImpl taskExecutor.execute() fail", e);

			}
		}
	}

	public class HandleRunner implements Runnable {
		private List<AsynItem> list;

		public HandleRunner(List<AsynItem> list) {
			this.list = list;
		}

		public void run() {
			try {

				Map<String, Object> params_map = new ConcurrentHashMap<String, Object>();
				List<Map<String, Object>> params_queue = new ArrayList<Map<String, Object>>();
				for (int i = 0; i < list.size(); i++) {
					AsynItem item = list.get(i);
					if (AsynItem.TYPE_MAP.equals(item.getType())) {
						params_map.put(item.getKey(), item.getObject());
					} else if (AsynItem.TYPE_QUEUE.equals(item.getType())) {
						Map<String, Object> map = new HashMap<String, Object>();
						map.put(item.getKey(), item.getObject());
						params_queue.add(map);
					}
				}

				if (params_map.size() > 0) {
					redis.setBatchSync(params_map);
				}
				if (params_queue.size() > 0) {
					redis.pushBatchSync(params_queue);
				}

			} catch (Throwable t) {
				logger.error("RedisHandlerImpl taskExecutor.execute() fail", t);
			}

		}

	}

	public void afterPropertiesSet() throws Exception {
		new Thread(this, "RedisHandlerImplServer").start();
		if (logger.isInfoEnabled()) {
			logger.info("启动Redis(RedisHandlerImplServer)服务！");
		}

	}

	public void setRedis(Redis redis) {
		this.redis = redis;
	}

	public void setTaskExecutor(TaskExecutor taskExecutor) {
		this.taskExecutor = taskExecutor;
	}
}