/*
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 * $Id: MySQLInvertedIndexRegistrar.java,v 1.5 2006/02/12 15:31:42 rampil Exp $
 */
package org.logical_paradox.koike.rss.index;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.commons.codec.binary.Base64;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.common.sql.DbUtils;
import org.logical_paradox.common.util.ExceptionUtils;
import org.logical_paradox.common.util.StringUtils;
import org.logical_paradox.koike.core.indexer.IndexAccessException;
import org.logical_paradox.koike.core.parser.Term;
import org.logical_paradox.rss.util.DocumentIdGenerator;

/**
 * 転置インデックスを登録するスレッド
 * パーザーが稼動している間に動作する予定．
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.5 $
 */
public class MySQLInvertedIndexRegistrar extends Thread {
	/** ロガー */
	private static final Log log = LogFactory.getLog(MySQLInvertedIndexRegistrar.class);
	/** 登録対象索引語辞書 */
	private HashMap<Long, HashMap<String, String>> queue = new HashMap<Long, HashMap<String, String>>();
	/** データベース接続 */
	private final Connection con;
	/** base64 encoder/decoder */
	private final Base64 base64 = new Base64();


	/**
	 * コンストラクタ．
	 * @param c データベース接続
	 */
	public MySQLInvertedIndexRegistrar(Connection c) {
		con = c;
		setDaemon(true);
	}
	/**
	 * スレッドメイン．
	 */
	public void run() {
		try {
			while(true) {
				synchronized(queue) {
					while(queue.size() == 0) {
						queue.wait();
					}
					try {
						// 登録(転送処理)
						register();
					} catch(Exception iae) {
						// いくつかの索引語の登録に失敗したが，可能な限り処理を続行する
						// ここは無視しておく
						log.warn("索引語の登録中に例外が発生しました:", iae);
					}
				}
			}
		} catch(InterruptedException e) {
			// スレッドの停止
		}
	}
	/**
	 * 索引語をキューに登録する．
	 * キューに登録されると，適切なタイミングで登録処理が起動する．
	 * @param digest 文書を特定するダイジェスト
	 * @param term 索引語の配列
	 */
	protected void add(String digest, Term[] term) throws IndexAccessException {
		synchronized(queue) {
			// レコード番号(document番号)に対応するキューを取り出す
			// キューが存在しない場合，新しく作成する
			long recordNo = DocumentIdGenerator.getDocumentNo(digest);
			Long queueId = new Long(recordNo);
			HashMap<String, String> recordQueue = queue.get(queueId);
			if(recordQueue == null) {
				recordQueue = new HashMap<String, String>();
			}
			for(int i = 0; i < term.length; i++) {
				String locations = (String)recordQueue.get(term);
				if(locations == null) {
					locations = "";
				}
				for(Iterator it = term[i].iterator(); it.hasNext();) {
					Integer offset = (Integer)it.next();
					if(StringUtils.isEmpty(locations) == false) {
						locations = locations + ",";
					}
					locations += offset;
					recordQueue.put(term[i].getTerm(), locations);
				}
			}
			queue.put(queueId, recordQueue);
			// キューで待っているスレッドを起こす
			queue.notifyAll();
		}
	}
	/**
	 * 索引語をデータベースに登録する．
	 */
	protected void register() throws IndexAccessException {
		// キューをコピーする
		HashMap map = null;
		synchronized(queue) {
			map = (HashMap)queue.clone();
			for(Iterator it = queue.keySet().iterator(); it.hasNext();) {
				Long qid = (Long)it.next();
				HashMap rq = (HashMap)queue.get(qid);
				if(rq != null) {
					map.put(qid, rq.clone());
					rq.clear();
				}
			}
			queue.clear();
		}

		for(Iterator it = map.keySet().iterator(); it.hasNext();) {
			Long qid = (Long)it.next();
			HashMap rq = (HashMap)map.get(qid);
			if(rq == null || rq.size() == 0) {
				// 送信キューが存在しない場合は何もしない
				continue;	
			}

			// データベースに転置インデックスを登録するSQLの構築
			// ある一定数でやめとく
			StringBuffer sb = new StringBuffer();
			sb.append("insert into inverted_index values");
			int length = rq.keySet().size();

			for(int i = 0; i < length; i++) {
				if(i > 0) {
					sb.append(",");
				}
				sb.append("(?,?,?)");
			}
			PreparedStatement pstmt = null;
			try {
				pstmt = con.prepareStatement(sb.toString());
			} catch(SQLException se) {
				// リソースの回収
				DbUtils.closeStatement(pstmt);
				pstmt = null;
				log.error("登録処理中にエラーが発生しました:" + ExceptionUtils.stackTraceToString(se), se);
				return;
			}

			try {
				int paramCnt = 1;
				for(Iterator qit = rq.keySet().iterator(); qit.hasNext();) {
					String key = (String)qit.next();
					String value = (String)rq.get(key);
	
					// 索引語
					String term = new String(base64.encode(key.getBytes()));
					pstmt.setString(paramCnt++, term);
					// レコード番号
					pstmt.setLong(paramCnt++, qid.longValue());
					// 位置情報
					pstmt.setString(paramCnt++, value);
				}

				// 引数の設定が完了したので一度にcommitする
				pstmt.executeUpdate();
				pstmt.close();

				// このドキュメントの送信キューをクリアする
				rq.clear();
			} catch(Exception se) {
				// 登録に失敗した場合，単に登録を無視する
				log.warn("登録に失敗: " + ExceptionUtils.stackTraceToString(se), se);
			} finally {
				// リソースの回収
				DbUtils.closeStatement(pstmt);
			}
		}
		// 送信キュー全体をクリアする
		map.clear();
	}
	/**
	 * スレッドを停止する．
	 */
	protected void shutdown() {
		interrupt();
	}
}
