package com.uas.search.console.b2b.jms;

import java.sql.SQLException;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.Session;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONException;
import com.uas.search.console.b2b.model.ParsedQueueMessage;
import com.uas.search.console.b2b.service.IndexService;
import com.uas.search.console.b2b.util.SearchConstants;

import oracle.jms.AQjmsAdtMessage;
import oracle.jms.AQjmsFactory;
import oracle.jms.AQjmsSession;

/**
 * 对数据库的消息队列进行实时监听
 * 
 * @author sunyj
 * @since 2016年11月9日 上午11:29:33
 */
@Service
public class AQListener {

	@Autowired
	private IndexService indexService;

	@Autowired
	private QueueMessageParser queueMessageParser;

	@Autowired
	private DruidDataSource dataSource;

	// 消息队列的消费者
	private MessageConsumer consumer;

	private Logger logger = LoggerFactory.getLogger(getClass());

	/**
	 * 开启实时更新索引
	 * 
	 * @param waitInterval
	 *            每次接收到jms消息后等待的时间（秒）
	 * @return 开启成功与否的提示信息
	 */
	public String start(final Long waitInterval) {
		String message = "";
		if (isRunning()) {
			message = "已存在运行的索引实时更新服务";
			logger.warn(message);
			return message;
		}

		try {
			QueueConnectionFactory queueConnectionFactory = AQjmsFactory.getQueueConnectionFactory(dataSource.getUrl(),
					new Properties());
			QueueConnection connection = queueConnectionFactory.createQueueConnection(dataSource.getUsername(),
					dataSource.getPassword());
			AQjmsSession session = (AQjmsSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Queue queue = session.getQueue(dataSource.getUsername(), SearchConstants.LUCENE_QUEUE_NAME);
			consumer = session.createConsumer(queue, null, QueueMessageTypeFactory.getFactory(), null, false);

			// 添加监听器，队列中一旦有消息入队，就会接受该消息（并不是真的实时，一般会有10秒以内的延迟）
			consumer.setMessageListener(new MessageListener() {
				@Override
				public void onMessage(Message message) {
					try {
						if (waitInterval != null) {
							// 等待waitInterval秒，为了等待数据表变动的事务提交
							// 如果时间过短，事务可能还未提交
							// 如果时间过长，实时更新速度很慢
							Thread.sleep(waitInterval * 1000);
						}
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
					AQjmsAdtMessage adtMessage = (AQjmsAdtMessage) message;
					try {
						QueueMessageTypeFactory payload = (QueueMessageTypeFactory) adtMessage.getAdtPayload();
						// 对出队的消息进行解析、处理
						process(payload.getMessage());
					} catch (JMSException e) {
						e.printStackTrace();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
			});
			connection.start();
			message = "索引实时更新服务成功开启";
			logger.info(message);
		} catch (JMSException e) {
			message = "索引实时更新服务开启失败";
			logger.error(message);
			e.printStackTrace();
			try {
				consumer.close();
			} catch (JMSException e1) {
				e1.printStackTrace();
			}
			consumer = null;
		}
		return message;
	}

	/**
	 * 关闭实时更新索引服务
	 * 
	 * @return 关闭成功与否的提示信息
	 */
	public String stop() {
		String message = "";
		if (!isRunning()) {
			message = "索引实时更新服务未开启或已关闭";
			logger.warn(message);
		} else {
			try {
				consumer.close();
				message = "索引实时更新服务成功关闭";
				logger.info(message);
			} catch (JMSException e) {
				message = "索引实时更新服务关闭失败";
				logger.error(message);
				e.printStackTrace();
			}
			consumer = null;
		}
		return message;
	}

	/**
	 * @return 索引实时更新服务是否正在运行
	 */
	public boolean isRunning() {
		return consumer != null;
	}

	/**
	 * 对得到的队列消息进行解析，之后根据解析出来的对象，对lucene索引进行添加、更新或删除操作
	 * 
	 * @param message
	 */
	private void process(String message) {
		ParsedQueueMessage parsedQueueMessage = null;
		logger.info(message);
		try {
			parsedQueueMessage = queueMessageParser.parse(message);
		} catch (JSONException e) {
			e.printStackTrace();
		}

		if (parsedQueueMessage == null) {
			logger.error("message parsing failed!");
			return;
		}
		indexService.maintainIndexes(parsedQueueMessage);
	}
}
