/*
 * $Id: ContentsSucker.java,v 1.2 2006/01/26 16:38:09 rampil Exp $
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 *
 */
package org.logical_paradox.rss.oqs4r.tools;

import java.rmi.RemoteException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.ResourceBundle;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.rss.dsr.RSSDistributedServiceRegistry;
import org.logical_paradox.rss.lookup.Lookup;
import org.logical_paradox.rss.lookup.LookupFactory;
import org.logical_paradox.rss.oqs4r.OQSConnection;
import org.logical_paradox.rss.oqs4r.ObjectQueueServer;
import org.logical_paradox.rss.util.DocumentIdGenerator;

/**
 * mysqlデータベースからコンテンツを取得して，oqs4rに登録する
 * i-frontのテスト用
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.2 $
 */
public class ContentsSucker {
	public static final String SUCKER_PROPERTY_PFX = "sucker";
	private static final Log log = LogFactory.getLog(ContentsSucker.class);
	private final ResourceBundle resource;
	private OQSConnection queue;
	private Connection con;
	private Statement stmt;
	private String sql;

	/**
	 * constructor.
	 * @param rb resource
	 */
	public ContentsSucker(ResourceBundle rb) throws Exception {
		resource = rb;
		init();
	}
	protected void init() throws Exception {
		// connect to oqs4r and get a connection
		Lookup lookup = LookupFactory.getLookup(resource.getString("dsr.uri"));
		ObjectQueueServer oqs4r = (ObjectQueueServer)lookup.lookup(RSSDistributedServiceRegistry.RSS_SERVID_OQS4R);
		queue = oqs4r.getConnection();

		// connect to mysql database server
		try {
			Class.forName(resource.getString("jdbc.driver.class"));
			con = DriverManager.getConnection(resource.getString("jdbc.connect.str"));
			stmt = con.createStatement();
			sql = resource.getString("jdbc.listing.sql");
		} catch(Exception e) {
			log.error("mysql database serverへの接続処理中にエラーが発生しました:", e);
			close();
		}
	}
	public void suck() throws InterruptedException {
		ListingThread thread = new ListingThread(con, queue);
		thread.start();
		thread.join();
	}
	public void close() throws RemoteException, SQLException {
		// close connection of oqs4r
		if(queue != null) {
			queue.close();
		}
		// close connection of mysql server
		if(stmt != null) {
			stmt.close();
		}
		if(con != null) {
			con.close();
		}
	}

	class ListingThread extends Thread {
		private final OQSConnection queue;

		/**
		 * constructor.
		 * @param c mysql server connection.
		 * @param q oqs4r connection
		 */
		protected ListingThread(Connection c, OQSConnection q) {
			con = c;
			queue = q;
			setDaemon(true);
		}
		public void run() {
			ResultSet rs = null;
			int enqueueCount = Integer.parseInt(resource.getString("enqueue.count.per.time"));
			long enqueueInterval = Long.parseLong(resource.getString("enqueue.interval.msec"));

			try {
				// listing
				rs = stmt.executeQuery(sql);
				int count = 0;
				List<String> list = new ArrayList<String>();
				while(rs.next()) {
					long cid = rs.getLong("cid");
					String digest = rs.getString("url");
					String docid = DocumentIdGenerator.getDocumentId(cid, digest);

					list.add(docid);
					count++;
					if(count < enqueueCount) {
						continue;
					}
					// enqueue
					count = 0;
					queue.add((String[])list.toArray(new String[0]));
					list.clear();

					// sleep
					Thread.sleep(enqueueInterval);
				}
				if(list.isEmpty() == false) {
					queue.add((String[])list.toArray(new String[0]));
				}
			} catch(SQLException e) {
				log.error("コンテンツリストの取得に失敗しました:", e);
				if(rs != null) {
					try {
						rs.close();
					} catch(Exception ee) {}
				}
			} catch (InterruptedException e) {
				log.error("待ち合わせ中に例外が発生しました:", e);
			} catch (RemoteException e) {
				log.error("oqs4rとの通信中に例外が発生しました:", e);
			}

			log.info("スレッドを停止しました");
		}
	}





	public static final void main(String[] args) throws Exception {
		ResourceBundle rb = ResourceBundle.getBundle(SUCKER_PROPERTY_PFX);
		ContentsSucker sucker = new ContentsSucker(rb);
		try {
			sucker.suck();
		} finally {
			sucker.close();
		}
	}
}
