package org.opengion.fukurou.queue;

import java.util.ArrayList;
import java.util.List;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * MQメッセージ受信用クラス。
 *
 * @og.group メッセージ連携
 *
 * @og.rev 5.10.15.2 (2019/09/20) 新規作成
 * 
 * @version 5
 * @author oota
 * @since JDK7
 */
public class QueueReceive_MQ implements QueueReceive{

	private QueueConnection connection = null;
	private QueueSession session = null;
	private QueueReceiver receiver = null;
	List<QueueReceiver> listReceiver = null;
	private boolean batch = false;
	
	/**
	 * 接続処理
	 * メッセージキューサーバに接続します。
	 * 
	 *  @param jmsServer jsmサーバ
	 *  @param sqsAccessKey sqs用awsアクセスキー(MQでは利用しません)
	 *  @param sqsSecretKey sqs用awsシークレットキー(MQでは利用しません)
	 */
	public void connect(final String jmsServer, final String sqsAccessKey, final String sqsSecretKey) {
		connect(jmsServer);
	}
	
	/**
	 * 接続処理
	 * jmsServerに接続します。
	 * MQの場合は、受信リスナーを設定して、随時メッセージ受信処理を行います。
	 * SQSの場合は最大受信件数の10件の処理を行います。
	 * 
	 * @param jmsServer 接続先情報 MQ:jndi接続先 SQS:URL
	 */
	private void connect(final String jmsServer) {	
		try {
			if(batch) {
				// バッチ用
				String mqUserId = System.getProperty("mqUserId");
				String mqPassword = System.getProperty("mqPassword");
				QueueConnectionFactory factory = new ActiveMQConnectionFactory(jmsServer);		
				connection = factory.createQueueConnection(mqUserId,  mqPassword);
			}else {
				// jndi接続用
				Context ctx = new InitialContext();
				QueueConnectionFactory factory = (QueueConnectionFactory)ctx.lookup("java:comp/env/" + jmsServer);
				connection = factory.createQueueConnection();
			}
			
			connection.start();
			
			// Receiveの作成
			session = connection.createQueueSession(false, QueueSession.CLIENT_ACKNOWLEDGE);
			
			// 初期化
			listReceiver = new ArrayList<QueueReceiver>();
		}catch(Exception e) {
			throw new RuntimeException("MQサーバの接続に失敗しました。：" + e.getMessage());
		}
	}

	/**
	 * 受信処理
	 * メッセージキューの受信の処理を行います。
	 * 
	 */
	@Override
	public QueueInfo receive(final String queueName) {
		QueueInfo queueInfo = null;
		
		try {
			Queue queue = session.createQueue(queueName);
			receiver = session.createReceiver(queue);
			
			TextMessage msg = (TextMessage)receiver.receive(1000);
			
			if(msg != null) {
				// メッセージ受信の確認応答
				msg.acknowledge();
				
				// メッセージの設定
				queueInfo = new QueueInfo();
				queueInfo.setMessage(msg.getText());
			}
		}catch(Exception e) {
			throw new RuntimeException(e.getMessage());
		}finally {
			try {
				receiver.close();
			}catch(Exception e) {}
		}
		
		return queueInfo;
	}
	
	/**
	 * リスナーの起動
	 * 指定したキュー名に対して、
	 * MessageListenerのリスナーを設定します。
	 * 
	 * @param queueName キュー名
	 * @param listener MessageListerを実装したクラス
	 */
	@Override
	public void setListener(final String queueName, MessageListener listener) {
		QueueReceiver receiver = null;
		try {	
			Queue queue = session.createQueue(queueName);
			receiver = session.createReceiver(queue);
			receiver.setMessageListener(listener);
			
			// リスナーの起動
			listReceiver.add(receiver);
		}catch(JMSException e) {
			throw new RuntimeException("リスナーの起動に失敗しました。" + e.getMessage());
		}
	}

	/**
	 * クローズリスナー
	 * レシーバーをクローズすることで、
	 * リスナーの処理を終了します。
	 */
	public void closeListener() {
		for(QueueReceiver receiver: listReceiver) {
			try {
				receiver.close();
			}catch(Exception e) {
				
			}
		}
		
		// 初期化
		listReceiver = null;
		listReceiver = new ArrayList<QueueReceiver>(); 
	}
	
	/**
	 * クローズ処理
	 * クローズ処理を行います。
	 */
	@Override
	public void close() {
		if(receiver != null) {
			try {
				receiver.close();
			}catch(Exception e) {
				
			}
		}
		if(session != null) {
			try {
				session.close();
			}catch(Exception e) {
				
			}
		}
		if(connection != null) {
			try {
				connection.close();
			}catch(Exception e) {
				
			}
		}
	}

	/**
	 * バッチ処理判定フラグを設定します。
	 * 
	 * @param batchFlg バッチ処理判定フラグ
	 */
	public void setBatchFlg(final Boolean batchFlg) {
		batch = batchFlg;
	}
	
	/**
	 * 検証用メソッド
	 * テスト用のメソッドです。
	 * 
	 * @param args 引数
	 */
	public static void main(String[] args) {
		QueueReceive receive = new QueueReceive_MQ();
		final String jmsServer = "tcp://localhost:61616";
		
		// バッチフラグにtrueを設定
		// 未設定の場合は、tomcatのjndi接続処理が実行されます。
		receive.setBatchFlg(true);
		
		// 認証情報の設定
		System.setProperty("mqUserId", "admin");
		System.setProperty("mqPassword", "admin");
		
		// 接続
		receive.connect(jmsServer, null, null);
		
		// 処理対象のキュー名
		String queueName = "queue01";
		
		
		// ** １件受信する場合
		QueueInfo queueInfo = receive.receive(queueName);
		if(queueInfo != null) {
			System.out.println("message:" + queueInfo.getMessage());	
		}else {
			System.out.println("キューが登録されていません。");
		}
		
//		// ** リスナーを設定して、受信を検知すると処理を実行します。(MQのみ)
//		// MessageListerを実装した、QueueReceiveListenerクラスを作成します。
//		MessageListener listener = new QueueReceiveListener();
//		receive.setListener(queueName, listener);
//		// 複数のキューにリスナーを設定することも可能です。
//		receive.setListener("queue02", listener);
//		
//		try {
//			// 1分間リスナーを起動しておく場合の、プロセス待機処理
//			Thread.sleep(60 * 1000);
//		}catch(InterruptedException e) {
//			throw new RuntimeException(e.getMessage());
//		}
		
		// リスナー利用時は、closeListenerを実行して、解放してください。
		receive.closeListener();
		
		// 終了処理
		receive.close();
	}
	
	/**
	 * QueueReceiveリスナークラス
	 * リスナー用のクラスです。
	 * MQに設定することで、メッセージが受信されると、
	 * 自動的にonMessageメソッドが実行されます。
	 *
	 */
	static class QueueReceiveListener implements MessageListener {
		/**
		 * メッセージ受信処理
		 * MQサーバにメッセージが受信されると、
		 * メソッドの処理が行われます。
		 * 
		 * @param Message 受信メッセージ
		 */
		@Override
		public void onMessage(final Message message) {

			// メッセージ受信
			TextMessage msg = (TextMessage) message;
			String msgText = "";

			try {
				// キューサーバのメッセージを取得
				msgText = msg.getText();
				// メーッセージの受信応答を返します。
				msg.acknowledge();
				
				System.out.println("message:" + msgText);

			} catch (JMSException e) {
				throw new RuntimeException(e.getMessage());
			}
		}
	}

}
