/*
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 * $Id: DbmysqlContentsAccessor.java,v 1.14 2006/02/22 15:47:01 akabane Exp $
 */
package org.logical_paradox.rss.rcm.accessor.mysql;

import java.io.ByteArrayInputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.common.charset.CharsetUtils;
import org.logical_paradox.common.sql.DbUtils;
import org.logical_paradox.common.util.ExceptionUtils;
import org.logical_paradox.rss.http.WebContents;
import org.logical_paradox.rss.rcm.accessor.ContentsAccessor;
import org.logical_paradox.rss.rcm.accessor.mysql.validator.ConnectionValidation;
import org.logical_paradox.rss.rcm.accessor.mysql.validator.ConnectionValidator;
import org.logical_paradox.rss.rcm.accessor.simple.VirtualQueueServer;
import org.logical_paradox.rss.rcm.df.DataFilter;
import org.logical_paradox.rss.rcm.df.DataFilterFactory;
import org.logical_paradox.rss.util.DocumentIdGenerator;

/**
 * DbmysqlContentsAccessor
 * mysqlデータベースをコンテンツストレージとして使用するアクセスインターフェース
 * 
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.14 $
 */
public class DbmysqlContentsAccessor implements ContentsAccessor {
	/** ロガー */
	private static final Log log = LogFactory.getLog(DbmysqlContentsAccessor.class);

	/** 接続先DSR */
	private String dsruri;
	/** コンテンツエンコーディング */
	private String contentsEncoding;
	/** コンテンツを格納する際，エンコーディング変換を必要とするかどうか(true:する / false:しない) */
	private boolean needToEncode = false;
	/** このインターフェースが接続状態かどうか(true:接続状態 / false:非接続状態) */
	private boolean activeFlag = false;
	/** 仮想キューサーバー */
	private VirtualQueueServer queue = null;
	/** データフィルタ */
	private static DataFilter dataFilter = null;

	/** データベース接続 */
	private Connection con;
	/** ステートメント */
	private PreparedStatement pstmt;
	/** データベース接続検証プロセス */
	private ConnectionValidator validator;

	/** JDBCドライバクラス名 */
	public static final String JDBC_DRIVER_CLASSNAME = "com.mysql.jdbc.Driver";
	/** データベース接続監視間隔(msec) */
	public static final long CONNECTION_WATCH_INTERVAL = 120 * 1000;

	/*
	 * データフィルタなどの初期化
	 */
	static {
		dataFilter = DataFilterFactory.getFilter(DataFilterFactory.GZIP);
	}

	/**
	 * コンストラクタ．
	 */
	public DbmysqlContentsAccessor() {
		activeFlag = false;
	}
	/**
	 * 初期化時に呼ばれる．
	 * @param initparam 初期化パラメータ
	 */
	public void init(String initparam) {
		open(initparam);
	}
	/**
	 * コンテンツストレージと接続する
	 * @param openstr 接続文字列
	 */
	public synchronized void open(String openstr) {
		if(isActive()) {
			// 既にオープンしている
			return;
		}

		if(openstr == null || openstr.trim().length() == 0) {
			throw new IllegalArgumentException();
		}

		// JDBCドライバのロードと接続
		try {
			activeFlag = true;
			// 仮想キューサーバーへ接続する
			queue = new VirtualQueueServer(dsruri);
			// データベース接続を作成する
			Class.forName(JDBC_DRIVER_CLASSNAME);
			con = DriverManager.getConnection(openstr);
			// ステートメントの作成
			String sql = "select * from contents where mkey=?";
			pstmt = con.prepareStatement(sql);

			// 接続監視スレッドを起動する
			validator = new ConnectionValidator(CONNECTION_WATCH_INTERVAL);
			ConnectionValidation cv = new ConnectionValidation(con, "select count(*) from contents");
			validator.add(cv);
			validator.start();
		} catch(Exception e) {
			// 何らかのエラーが生じた場合，リソースの回収を行なう
			if(con != null) {
				DbUtils.closeConnection(con);
				con = null;
			}
			if(pstmt != null) {
				DbUtils.closeStatement(pstmt);
				pstmt = null;
			}
			log.error("インターフェースの初期化に失敗しました:" + ExceptionUtils.stackTraceToString(e));
			throw new IllegalArgumentException(e.getMessage());
		}
	}

	/**
	 * コンテンツを登録する
	 * @param contents 登録するコンテンツ
	 */
	public synchronized void regist(WebContents contents) {
		String sql = "insert into contents values(null,?,?,?,?,?)";
		String url = contents.getLocation();
		String document = contents.getDocument();

		byte[] barray = new byte[0];
		try {
			// エンコーディング変換が必要であれば変換する．
			if(needToEncode) {
				log.info("コンテンツのエンコーディングを変換しています[" + contentsEncoding + "]");
				document = CharsetUtils.encode(contentsEncoding, document);
				barray = dataFilter.compressContents(document.getBytes("ISO-8859-1"));
			} else {
				barray = dataFilter.compressContents(document.getBytes());
			}
		} catch (Exception e1) {
			// コンテンツの圧縮に失敗した場合，もう登録できないのでワーニングを出して無視する
			log.warn("コンテンツのフィルタリングに失敗しました:" + ExceptionUtils.stackTraceToString(e1), e1);
			return;
		}
		ByteArrayInputStream bi = new ByteArrayInputStream(barray);
		String sitename = contents.getSitename();
		String title = contents.getTitle();
		PreparedStatement updateStmt = null;
		Statement st = null;
		ResultSet rs = null;
		try {
			// 指定URLのMD5ダイジェストを取得して，コンテンツと関連付けるキーとする
			String digest = dataFilter.filterURL(url);
			updateStmt = con.prepareStatement(sql);
			updateStmt.setString(1, digest);						// mkey
			updateStmt.setString(2, sitename);					// sitename
			updateStmt.setString(3, title);						// title
			updateStmt.setString(4, url);						// URL
			updateStmt.setBinaryStream(5, bi, barray.length);	// contents
			updateStmt.executeUpdate();
			updateStmt.close();

			// 生成されたレコードのIDを取得する
			sql = "select last_insert_id() as did from contents";
			st = con.createStatement();
			rs = st.executeQuery(sql);
			if(rs.next()) {
				long did = rs.getLong("did");
				// キューにURLを投入する
				String documentId = DocumentIdGenerator.getDocumentId(did, digest);
				queue.enqueue(documentId);
			} else {
				log.warn("何らかの原因でドキュメント番号を発番できませんでした");
			}
		} catch(Exception e) {
			// 登録できなかった
			log.warn("コンテンツの登録に失敗しました:" + ExceptionUtils.stackTraceToString(e), e);
		} finally {
			// 使用したリソースの回収
			if(updateStmt != null) {
				DbUtils.closeStatement(updateStmt);
			}
			if(rs != null) {
				DbUtils.closeResultSet(rs);
			}
			if(st != null) {
				DbUtils.closeStatement(st);
			}
		}
	}

	/**
	 * コンテンツを削除する．
	 * @param url 削除対象のコンテンツ
	 */
	public void unregist(String url) {
		if(isActive() == false || url == null || url.trim().length() == 0) {
			throw new IllegalStateException();
		}

		String sql = "delete from contents where mkey=?";
		PreparedStatement deleteStmt = null;
		Connection con = null;
		try {
			deleteStmt = con.prepareStatement(sql);
			deleteStmt.setString(1, url);
			deleteStmt.executeUpdate(sql);
		} catch(SQLException se) {
			// 削除できなかった
			log.warn("コンテンツの削除に失敗しました[" + url + "]: " + ExceptionUtils.stackTraceToString(se), se);
		} finally {
			// リソースの回収
			if(deleteStmt != null) {
				DbUtils.closeStatement(deleteStmt);
				deleteStmt = null;
			}
		}
	}

	/**
	 * 全てのコンテンツを削除する
	 */
	public void unregistAll() {
		if(isActive() == false) {
			throw new IllegalStateException();
		}

		String sql = "delete from contents";
		Statement st = null;
		Connection con = null;
		try {
			st = con.createStatement();
			st.executeUpdate(sql);
		} catch(SQLException se) {
			// 削除できなかった
			log.warn("コンテンツの削除に失敗しました:" + ExceptionUtils.stackTraceToString(se), se);
		} finally {
			// リソースの回収
			if(st != null) {
				DbUtils.closeStatement(st);
				st = null;
			}
		}
	}

	/**
	 * 指定URLに対応するコンテンツを返す
	 */
	public String find(String url) {
		if(isActive() == false || url == null || url.trim().length() == 0) {
			throw new IllegalStateException();
		}

		String contents = null;
		ResultSet rs = null;
		try {
			pstmt.setString(1, url);
			rs = pstmt.executeQuery();
			if(rs.next()) {
				contents = rs.getString(1);
			}
		} catch(SQLException se) {
			// エラーの場合は何もしない
			log.warn("コンテンツの取得に失敗しました[" + url + "]:" + ExceptionUtils.stackTraceToString(se), se);
		} finally {
			// リソースの回収
			if(rs != null) {
				DbUtils.closeResultSet(rs);
				rs = null;
			}
		}
		return contents;
	}

	/**
	 * ストレージとの接続を閉じる．
	 * 何が何でもクローズする．
	 */
	public synchronized void close() {
		if(isActive() == false) {
			throw new IllegalStateException("connector has already been closed");
		}
		// DB接続のクローズ
		try {
			if(pstmt != null) {
				pstmt.close();
				pstmt = null;
			}
		} catch(SQLException e) {
			// どうしようもないので無視しておく
			log.warn("ステートメントクローズ処理中に例外が発生しました:" + ExceptionUtils.stackTraceToString(e), e);
		} finally {
			// どうでもいいけどDB接続だけはクローズしておく
			if(con != null) {
				try {
					con.close();
				} catch(SQLException e) {
					// もうどうでもいい
					log.warn("データベース接続終了時に例外が発生しました:" + ExceptionUtils.stackTraceToString(e), e);
				}
				con = null;
			}
		}
		activeFlag = false;
	}

	/**
	 * このインターフェースが使用可能かどうか返す
	 */
	protected boolean isActive() {
		return activeFlag;
	}

	/**
	 * ファイナライザ
	 */
	public void finalize() {
		try {
			close();
		} catch(Exception e) {
			// 何もしない．何もできない
		}
	}
	/**
	 * 接続先DSRを設定する．
	 * @param uri 接続先DSR uri
	 */
	public void setDSRURI(String uri) {
		dsruri = uri;
	}
	/**
	 * コンテンツのエンコーディングを設定する．
	 * @param encoding エンコーディング．
	 */
	public void setContentsEncoding(String encoding) {
		contentsEncoding = encoding;
		try {
			needToEncode = CharsetUtils.testNeedToEncode(encoding);
		} catch(Exception e) {
			needToEncode = false;
		}
	}
}
