package org.opengion.plugin.daemon;

import java.io.File;
import java.util.Date;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.opengion.fukurou.business.BizUtil;
import org.opengion.fukurou.queue.QueueInfo;
import org.opengion.fukurou.queue.QueueReceive;
import org.opengion.fukurou.queue.QueueReceiveFactory;
import org.opengion.fukurou.util.HybsTimerTask;
import org.opengion.fukurou.util.StringUtil;
import org.opengion.hayabusa.common.HybsSystem;
import org.opengion.hayabusa.common.HybsSystemException;
import org.opengion.hayabusa.queue.DBAccessQueue;

/**
 * メッセージキュー受信 メッセージキューの受信処理を行います。
 * 
 * @og.group メッセージ連携
 * 
 * @og.rev 5.10.15.2 (2019/09/20) 新規作成
 *
 * @version 5.0
 * @author oota
 * @since JDK7
 *
 */
public class Daemon_QueueReceive extends HybsTimerTask {
	private int loopCnt = 0;
	private QueueReceive queueReceive = null;

	private static final int LOOP_COUNTER = 24;

	private final String CLOUD_SQS_ACCESS_KEY = HybsSystem.sys("CLOUD_SQS_ACCESS_KEY");
	private final String CLOUD_SQS_SECRET_KEY = HybsSystem.sys("CLOUD_SQS_SECRET_KEY");
	private final String MQ_QUEUE_TYPE;
	private final String MQ_QURUE_SERVER_URL = HybsSystem.sys("MQ_QUEUE_SERVER_URL");
	private final String MQ_QUEUE_RECEIVE_LISTENER =HybsSystem.sys("MQ_QUEUE_RECEIVE_LISTENER"); 

	private final String SYSTEM_ID = HybsSystem.sys("SYSTEM_ID");
	private final String USER_ID = "CYYYYY";
	private final String PG_ID;
	private final String DMN_NAME = "QueueReceiveDMN";
	private final DBAccessQueue dbAccessQueue;

	/**
	 * コンストラクター
	 * 初期処理を行います。
	 */
	public Daemon_QueueReceive() {
		super();
		
		// パラメータの設定
		if(!StringUtil.isNull(HybsSystem.sys("MQ_QUEUE_TYPE"))) {
			MQ_QUEUE_TYPE = HybsSystem.sys("MQ_QUEUE_TYPE").toUpperCase();
			PG_ID = StringUtil.cut("QueRec" + MQ_QUEUE_TYPE, 10);
		}else {
			throw new RuntimeException("システムリソースにMQ_QUEUE_TYPEを登録して下さい");
		}

		dbAccessQueue = new DBAccessQueue(SYSTEM_ID, USER_ID, PG_ID, DMN_NAME);
	}

	/**
	 * 初期処理 MQサーバに接続します。
	 */
	@Override
	public void initDaemon() {
		// 開始ログO
		StringBuilder errMsg = new StringBuilder();
		if (MQ_QUEUE_TYPE == null) {
			errMsg.append("MQ_QUEUE_TYPE");
		}
		if (MQ_QURUE_SERVER_URL == null) {
			errMsg.append(" MQ_QUEUE_SERVER_URL");
		}

		if (errMsg.length() > 0) {
			errMsg.append(" キュータイプを特定するために、左記のシステムリソースを登録して下さい。");
			throw new HybsSystemException(errMsg.toString());
		}

		String queueType = MQ_QUEUE_TYPE.toUpperCase();

		// 開始ログ
		System.out.println("MQキュータイプ：" + queueType);
		System.out.println("MQサーバーURL：" + MQ_QURUE_SERVER_URL);

		queueReceive = QueueReceiveFactory.newQueueReceive(queueType);

		queueReceive.connect(MQ_QURUE_SERVER_URL, CLOUD_SQS_ACCESS_KEY, CLOUD_SQS_SECRET_KEY);

	}

	/**
	 * 開始処理 タイマータスクのデーモン処理の開始ポイントです。
	 */
	@Override
	protected void startDaemon() {
		if (loopCnt % LOOP_COUNTER == 0) {
			loopCnt = 1;
			System.out.println();
			System.out.print(toString() + " " + new Date() + " ");
		} else {
			// 対象 キュー名(グループ名)とbizlogic名の取得処理
			String[][] ge67vals = dbAccessQueue.setlectGE67();
			// キュー情報登録チェック
			if (ge67vals.length == 0) {
				String errMsg = "GE67にキュー情報が登録されていません。";
				throw new RuntimeException(errMsg);
			}
			// MQとSQSで処理を分岐
			// MQ:指定キューIDからキューメッセージを取得
			// SQS:キューメッセージを取得してからキューID(グループID)を取得
			switch (MQ_QUEUE_TYPE) {
				case "MQ":
					processMq(ge67vals);
					break;
				case "SQS":
					processSqs(ge67vals);
					break;
				default:
					String errMsg = "リソース(MQ_QUEUE_TYPE)の値が不正です。：" + MQ_QUEUE_TYPE;
					throw new RuntimeException(errMsg);
			}

			loopCnt++;
		}
	}

	/**
	 * MQ用の処理
	 * GE67に登録されているキューIDの、
	 * メッセージキューを取得して処理を行います。
	 * 
	 * @param ge67vals
	 */
	private void processMq(final String[][] ge67vals) {
		boolean listenerMode = false;
		
		if("true".equals(MQ_QUEUE_RECEIVE_LISTENER)) {
			listenerMode = true;
		}
		
		if(listenerMode) {
			// リスナーの初期化
			queueReceive.closeListener();
		}

		// ge67のキューリスト分繰り返します
		for (int row = 0; row < ge67vals.length; row++) {
			String queueId = ge67vals[row][0];
			String bizLogicId = ge67vals[row][1];

			if(listenerMode) {
				// リスナーを設定して、動的な受信処理(MQ専用)
				QueueReceiveListener listener = new QueueReceiveListener(queueId, bizLogicId);
				queueReceive.setListener(queueId, listener);
			}else {
				// １件の受信処理
				QueueInfo queueInfo = queueReceive.receive(queueId);
				if (queueInfo != null) {
					processMessage(queueId, bizLogicId, queueInfo.getMessage());
					// 1件処理を行ったら処理を終了します。
					break;
				}
			}
		}
	}

	/**
	 * SQS用の処理
	 * SQSはグループIDを指定して、キューを取得することはできず、
	 * 任意のキューを１つ取得してから、
	 * 判定処理を行います。
	 * GE67に登録されていないグループIDのキューが取得された場合は、
	 * GE68にエラーレコードを登録します。
	 * 
	 * @param ge67vals
	 */
	private void processSqs(final String[][] ge67vals) {
		// 下記はSQSの場合(キューを１件取得して処理)
		QueueInfo queueInfo = queueReceive.receive(null);
		
		// キューが未取得の場合
		if(queueInfo == null) {
			return;
		}
		
		// 受信したキューを処理
		final String groupId = queueInfo.getSqsFifoGroupId();
		Boolean existsFlg = false;
		// valsにグループIDのレコードが存在するか検索
		for (int row = 0; row < ge67vals.length; row++) {
			final String queueId = ge67vals[row][0];

			if (groupId != null && groupId.equals(queueId)) {
				// 該当レコードあり
				final String bizLogicId = ge67vals[row][1];
				processMessage(queueId, bizLogicId, queueInfo.getMessage());

				existsFlg = true;
				break;
			}
		}

		if (!existsFlg) {
			// 該当groupIdの未登録エラー
			// 処理番号生成
			String syoriNo = dbAccessQueue.generateSyoriNo();
			dbAccessQueue.insertGE68(groupId, syoriNo, null, queueInfo.getMessage());
			dbAccessQueue.updateGE68Error(syoriNo, "SQSキューに設定されているグループIDが、GE67に未登録です。");
		}
	}

	/**
	 * キャンセル処理
	 * タイマータスクのデーモン処理の終了ポイントです。
	 */
	@Override
	public boolean cancel() {
		if (queueReceive != null) {
			queueReceive.close();
		}

		return super.cancel();
	}

	/**
	 * メッセージの処理
	 *  受信したメッセージをbizLogicに渡して、
	 *  処理を実行します。
	 *  
	 * @param queueId キューID
	 * @param bizLogicId bizLogicId
	 * @param msgText 受信メッセージ
	 */
	private void processMessage(final String queueId, final String bizLogicId, final String msgText) {
		String syoriNo = "";
		try {
			// 処理番号生成
			syoriNo = dbAccessQueue.generateSyoriNo();

			// 管理テーブル登録
			dbAccessQueue.insertGE68(queueId, syoriNo, bizLogicId, msgText);

			// bizLogicの処理を実行
			callActBizLogic(SYSTEM_ID, bizLogicId, msgText);

			// 管理テーブル更新(完了)
			dbAccessQueue.updateGE68(syoriNo, DBAccessQueue.FGKAN_END);

		} catch (Throwable te) {
			// bizLogicでのエラーはログの未出力して、処理を継続します。
			// bizLogicのエラー情報はCauseに格納されているため、Causeから取得します。
			String errMessage = null;
			if (te.getCause() != null) {
				// causeが設定されている場合のエラー情報
				errMessage = te.getCause().getMessage();
			} else {
				// causeが未設定の場合のエラー情報
				errMessage = te.getMessage();
			}
			System.out.println(errMessage);
			try {
				// エラーテーブルに登録
				dbAccessQueue.updateGE68Error(syoriNo, errMessage);
			} catch (Exception e) {
				// ここでのエラーはスルーします。
				System.out.println("管理テーブル登録エラー：" + e.getMessage());
			}
		}
	}

	/**
	 * bizLogic処理の呼び出し
	 * 必要なパス情報をリソースから取得して、
	 * BizUtil.actBizLogicにパス情報を渡すことで、
	 * bizLogicの処理を行います。
	 * 
	 * @param systemId  システムID
	 * @param logicName ロジックファイル名
	 * @param msgText   メッセージ
	 * @throws Throwable エラー情報
	 */
	private void callActBizLogic(String systemId, String logicName, String msgText) throws Throwable {
		// 対象 クラスパスの生成
		// HotDeploy機能を使用する場合に、Javaクラスをコンパイルするためのクラスパスを設定します。
		// 対象となるクラスパスは、WEB-INF/classes 及び WEB-INF/lib/*.jar です。
		// bizLogicTag.javaのコードを移植
		StringBuilder sb = new StringBuilder();
		sb.append('.').append(File.pathSeparatorChar);
		File lib = new File(HybsSystem.sys("REAL_PATH") + "WEB-INF" + File.separator + "lib");
		File[] libFiles = lib.listFiles();
		for (int i = 0; i < libFiles.length; i++) {
			sb.append(libFiles[i].getAbsolutePath()).append(File.pathSeparatorChar);
		}
		sb.append(HybsSystem.sys("REAL_PATH") + "WEB-INF" + File.separator + "classes").append(File.pathSeparatorChar);
		// bizの下のパス
		sb.append(HybsSystem.sys("REAL_PATH") + HybsSystem.sys("BIZLOGIC_CLASS_PATH")).append(File.pathSeparatorChar);
		// 上記で生成したクラスパスをclassPathに格納
		String classPath = sb.toString();

		// ソースパス情報の生成
		String srcDir = HybsSystem.sys("REAL_PATH") + HybsSystem.sys("BIZLOGIC_SRC_PATH");
		String classDir = HybsSystem.sys("REAL_PATH") + HybsSystem.sys("BIZLOGIC_CLASS_PATH");
		boolean isAutoCompile = HybsSystem.sysBool("BIZLOGIC_AUTO_COMPILE");
		boolean isHotDeploy = HybsSystem.sysBool("BIZLOGIC_HOT_DEPLOY");

		// bizLogicに渡すパラメータ
		String[] keys = new String[] { "message" };
		String[] vals = new String[] { msgText };

		// bizLogic処理の実行
		BizUtil.actBizLogic(srcDir, classDir, isAutoCompile, isHotDeploy, classPath, systemId, logicName, keys, vals);
	}

	/**
	 * 受信処理リスナー用のインナークラス
	 * QueueReceiveリスナークラス リスナー用のクラスです。
	 *  MQに設定することで、メッセージが受信されると、
	 * onMessageメソッドが実行されます。
	 *
	 */
	class QueueReceiveListener implements MessageListener {
		private String queueId = "";
		private String bizLogicId = "";

		/**
		 * コンストラクター 初期処理を行います。
		 * 
		 * @param quId  キューID
		 * @param bizId bizLogicId
		 */
		public QueueReceiveListener(final String quId, final String bizId) {
			queueId = quId;
			bizLogicId = bizId;
		}

		/**
		 * メッセージ受信処理 MQサーバにメッセージが受信されると、 メソッドの処理が行われます。
		 * 
		 * @param Message 受信メッセージ
		 */
		@Override
		public void onMessage(final Message message) {
			// 要求番号
			String ykno = "";

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

			try {
				// キューサーバのメッセージを取得
				msgText = msg.getText();

				// メーッセージの受信応答を返します。
				msg.acknowledge();

				processMessage(queueId, bizLogicId, msgText);

			} catch (JMSException jmse) {
				try {
					// 管理テーブル更新
					// 管理テーブル更新(エラー)
					dbAccessQueue.updateGE68(ykno, DBAccessQueue.FGKAN_ERROR);
				} catch (Exception e) {
					// ここでのエラーはスルーします。
					System.out.println("管理テーブル登録エラー：" + e.getMessage());
				}

				throw new HybsSystemException("bizLogicの処理中にエラーが発生しました。" + jmse.getMessage());
			}
		}
	}
}
