/*
 * $Id: ObjectQueueServerImpl.java,v 1.7 2005/10/18 16:22:05 rampil Exp $
 * Copyright (c) 2004 LOGICAL-PARADOX.ORG
 */
package org.logical_paradox.rss.oqs4r;

import java.io.Serializable;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.common.util.Cache;
import org.logical_paradox.rss.dsr.RSSDistributedServiceRegistry;
import org.logical_paradox.rss.lookup.LookupFactory;

/**
 * オブジェクトキューサーバー for RSS
 * org.logical_paradox.commmon.oqsをRSSで使用できるようにする
 * 
 * @author  satoshi akabane@logical-paradox.org
 * @version $Revision: 1.7 $
 */
public class ObjectQueueServerImpl extends UnicastRemoteObject implements ObjectQueueServer {
	/** ロガー */
	private static final Log logger = LogFactory.getLog(ObjectQueueServerImpl.class);
	/** コンフィグファイル名 */
	public static final String OQS4R_CONFIG_FILENAME = "conf/oqs4r.conf";

	// 分散システム上での識別ID
	private String nodeId = null;
	// キューの有効期限を監視するスレッド
	private GC gc;

	// キューを管理するオブジェクト
	protected Cache queue;
	// 接続インターフェース
	protected OQSConnection[] preparedConnections = null;
	// 接続インターフェースのカウンタ
	protected int nextidx = 0;

	// キューシステムのコンフィグレーション
	protected OQS4RConfig config;

	/**
	 * コンストラクタ
	 * @throws RemoteException
	 */
	protected ObjectQueueServerImpl() throws RemoteException {
		queue = new Cache();
	}
	/**
	 * サーバーの開始
	 */
	public void start() throws Exception {
		 config = new OQS4RConfig(OQS4R_CONFIG_FILENAME);

		// 分散サービスとして登録する
		nodeId = LookupFactory.getLookup(config.getDSRName()).bind(RSSDistributedServiceRegistry.RSS_SERVID_OQS4R, this);

		int connections = config.getPreparedConnections();

		// 接続インターフェースの生成
		preparedConnections = new OQSConnection[connections];
		for(int i = 0; i < preparedConnections.length; i++) {
			preparedConnections[i] = new OQSConnectionImpl(this);
		}

		logger.info("Object Queue Server ['" + nodeId + "']を開始しました");


		// ガベージコレクタを生成して始動する
		gc = new GC(queue, config.getCacheExpireSec());
		gc.start();
	}

	/*
	 * キューの操作は同じパッケージのクラスからしかアクセスできない
	 */
	public Serializable next() throws Exception {
		return next(false);
	}
	public Serializable next(boolean blocking) throws Exception {

		if(blocking == false) {
			// 処理をブロックしない
			synchronized(queue) {
				return (Serializable)queue.getOne();
			}
		}

		// キューに何か積まれるまで処理をブロックする
		Serializable rc = null;
		synchronized(queue) {
			while(queue.size() == 0) {
				queue.wait();
			}
			rc = (Serializable)queue.getOne();
			queue.notifyAll();
		}

		logger.info("デキュー:" + rc.toString());
		return rc;
	}
	/**
	 * キューにオブジェクトを追加する
	 * @param o 追加するオブジェクト
	 * @throws Exception 追加に失敗した場合に発生する例外
	 */
	public void add(Serializable o) throws Exception {
		queue.add(o);
		logger.info("エンキュー:" + o.toString());
	}
	/**
	 * キューを返す
	 * @return キュー
	 */
	protected Cache getCache() {
		return queue;
	}
	/**
	 * 接続インターフェースを返す
	 * @return 接続インターフェース
	 */
	public OQSConnection getConnection() throws RemoteException {
		OQSConnection aConnection = preparedConnections[nextidx++];
		if(nextidx >= preparedConnections.length) {
			nextidx = 0;
		}
		return aConnection;
	}
	/**
	 * サービス名を返す
	 * @return サービス名
	 * @throws RemoteException サービス名取得に失敗
	 */
	public String getName() throws RemoteException {
		return nodeId;
	}
	/**
	 * 完全修飾サービス名を返す
	 * @return 完全修飾サービス名
	 * @throws RemoteException サービス名取得に失敗
	 */
	public String getFqsn() throws RemoteException {
		return nodeId + "@" + config.getDSRName();
	}
	/**
	 * サービスを停止する
	 * RMIサーバーとして実装されている場合，unexportされる
	 * @param force true:強制停止 / false:安全に停止
	 * @throws RemoteException サービス停止に失敗
	 */
	public void shutdown(boolean force) throws RemoteException {
		logger.info("サーバーを停止します");

		// GCを停止する
		gc.shutdown();

		// exportしている接続オブジェクトを全て回収する
		for(int i = 0; i < preparedConnections.length; i++) {
			if(preparedConnections[i] != null) {
				unexportObject(preparedConnections[i], force);
			}
			logger.info("接続オブジェクトを解放しました");
		}

		// DSRから切り離す
		try {
			logger.info("Object Queue ServerをDSRから切り離しています");
			LookupFactory.getLookup(config.getDSRName()).unbind(RSSDistributedServiceRegistry.RSS_SERVID_OQS4R, nodeId);
		} catch (Exception e1) {
			throw new RemoteException(e1.getMessage());
		}

		UnicastRemoteObject.unexportObject(this, force);

		logger.info("サーバーを停止しました");
	}
}
