package com.goafanti.message.queue;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.data.redis.core.RedisTemplate;

import com.goafanti.message.bo.MessageAdapter;

public class MessageRedisQueue implements InitializingBean, DisposableBean {
	private RedisTemplate<String, Object>		redisTemplate;
	private String								key;
	private RedisConnectionFactory				factory;
	private RedisConnection						connection;
	private BoundListOperations<String, Object>	listOperations;

	private Lock								lock	= new ReentrantLock();

	private RedisQueueListener<MessageAdapter>	listener;
	private Thread								listenerThread;

	private boolean								isClosed;

	public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
		this.redisTemplate = redisTemplate;
	}

	public void setListener(RedisQueueListener<MessageAdapter> listener) {
		this.listener = listener;
	}

	public void setKey(String key) {
		this.key = key;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		factory = redisTemplate.getConnectionFactory();
		connection = RedisConnectionUtils.getConnection(factory);

		listOperations = redisTemplate.boundListOps(key);
		if (listener != null) {
			listenerThread = new ListenerThread();
			listenerThread.setDaemon(true);
			listenerThread.start();
		}
	}

	/**
	 * blocking remove and get last item from queue:BRPOP
	 * 
	 * @return
	 */
	public Object takeFromTail(int timeout) throws InterruptedException {
		lock.lockInterruptibly();
		try {
			return listOperations.rightPop(timeout, TimeUnit.SECONDS);
		} finally {
			lock.unlock();
		}
	}

	public Object takeFromTail() throws InterruptedException {
		return takeFromTail(0);
	}

	/**
	 * 从队列的头，插入
	 */
	public void pushFromHead(Object value) {
		listOperations.leftPush(value);
	}

	public void pushFromTail(Object value) {
		listOperations.rightPush(value);
	}

	/**
	 * noblocking
	 * 
	 * @return null if no item in queue
	 */
	public Object removeFromHead() {
		return listOperations.leftPop();
	}

	public Object removeFromTail() {
		return listOperations.rightPop();
	}

	/**
	 * blocking remove and get first item from queue:BLPOP
	 * 
	 * @return
	 */
	public Object takeFromHead(int timeout) throws InterruptedException {
		lock.lockInterruptibly();
		try {
			return listOperations.leftPop(timeout, TimeUnit.SECONDS);
		} finally {
			lock.unlock();
		}
	}

	public Object takeFromHead() throws InterruptedException {
		return takeFromHead(0);
	}

	@Override
	public void destroy() throws Exception {
		if (isClosed) {
			return;
		}
		shutdown();
		RedisConnectionUtils.releaseConnection(connection, factory);
	}

	private void shutdown() {
		try {
			listenerThread.interrupt();
		} catch (Exception e) {
		}
	}

	private class ListenerThread extends Thread {
		@Override
		public void run() {
			try {
				while (true) {
					Object value = takeFromHead();
					// 逐个执行
					if (value != null) {
						try {
							listener.onMessage((MessageAdapter) value);
						} catch (Exception e) {
						}
					}
				}
			} catch (InterruptedException e) {
			}
		}
	}
}
