/*
 * $Id: RNodeClientFactoryImpl.java,v 1.15 2006/01/28 03:37:07 rampil Exp $
 * Copyright (c) 2005 LOGICAL-PARADOX.ORG
 */
package org.logical_paradox.rss.router;
import static org.logical_paradox.rss.dsr.RSSDistributedServiceRegistry.RSS_SERVID_ROUTING_NODE_CLIENT_FACTORY;

import java.net.MalformedURLException;
import java.rmi.NoSuchObjectException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.NoSuchElementException;
import java.util.Vector;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.common.cli.Argument;
import org.logical_paradox.common.cli.ArgumentParser;
import org.logical_paradox.rss.ModuleException;
import org.logical_paradox.rss.RSSSubModule;
import org.logical_paradox.rss.dsync.SyncCommand;
import org.logical_paradox.rss.dsync.SyncQueue;
import org.logical_paradox.rss.dsync.SyncQueueException;
import org.logical_paradox.rss.http.URLPool;
import org.logical_paradox.rss.lookup.LookupFactory;
import org.logical_paradox.rss.router.algorithm.AllocatingAlgorithm;
import org.logical_paradox.rss.router.algorithm.AllocationException;

/**
 * RNodeClientFactoryImpl(実装)
 * 経路決定ノードクライアントを作成するファクトリ<br>
 * 必要に応じて新しいインスタンスを追加する．必要ない場合は，既存のものを返す
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.15 $
 */
public class RNodeClientFactoryImpl extends UnicastRemoteObject implements RNodeClientFactory, RSSSubModule {
	/** シリアルバージョンUID */
	private static final long serialVersionUID = -1646722572391043386L;
	/** ロガー */
	private static final Log log = LogFactory.getLog(RNodeClientFactoryImpl.class);
	/** コンフィグファイル名 */
	public static final String RSS_ROUTING_NODE_CONFIG = "../conf/rssroutingfactory.conf";

	/** コンフィグファイル名 */
	private String rnodeConfigFilename = RSS_ROUTING_NODE_CONFIG;

	private Hashtable<String,RoutingNodeClient> searchNodes = null;	// このクライアントファクトリが管理している探索ノード
	private String factoryServiceName = null;							// このサービスの名前
	private RSSRoutingNodeConfig config = null;						// このファクトリのコンフィグ
	private URLPool unTracedURLs = null;								// 未走査URL
	private SyncQueue syncQueue = new SyncQueue();						// 探索ロック差分情報用のキュー
	private SyncQueue rejectedCommandQueue = new SyncQueue();			// 拒否されてたロック要求を保管するキュー
	private AllocatingAlgorithm allocater = null;						// 分配アルゴリズム
	private String DSRURL;

	/**
	 * コンストラクタ
	 * @throws RemoteException インスタンス化に失敗
	 */
	public RNodeClientFactoryImpl() throws RemoteException {
	}

	/**
	 * コンストラクタ
	 * @param filename コンフィグファイル名
	 * @throws RemoteException インスタンス化に失敗
	 */
	public RNodeClientFactoryImpl(String filename) throws RemoteException {
		rnodeConfigFilename = filename;
	}

	/**
	 * 初期化する
	 * @param dsrurl DSR
	 * @throws RemoteException 初期化に失敗した
	 */
	public void init(String dsrurl) throws RemoteException {
		DSRURL = dsrurl;
		if(DSRURL == null || DSRURL.trim().length() == 0) {
			throw new RemoteException("dsr service name is empty");
		}

		unTracedURLs = new URLPool();
		searchNodes = new Hashtable<String,RoutingNodeClient>();

		try {
			// 経路決定ノード選択子にサービスインをする
			log.info("分散サービスレジストリ'" + DSRURL + "'にサービスインしています");

			factoryServiceName = LookupFactory.getLookup(DSRURL).bind(RSS_SERVID_ROUTING_NODE_CLIENT_FACTORY, this);
			log.info("経路決定ノード: " + getNodeId());

			log.info("サービスインに成功しました");
		} catch(Exception e) {
			log.info("サービスインに失敗しました");
			throw (RemoteException)e;
		}
	}
	/**
	 * このファクトリの名前を返す
	 */
	public String getNodeId() throws RemoteException {
		return factoryServiceName;
	}

	/**
	 * 探索ノードをバインドする
	 */
	public RoutingNodeClient bindSearchNode() throws RemoteException {
		log.info("探索ノードをバインドしています");

		// 探索ノードをバインドする処理
		RoutingNodeClient client = attachSearchNode();

		try {
			log.info("探索ノードが，名前'" + client.getNodeId() + "'として登録されました");
		} catch(Exception e) {
			throw new RemoteException(e.getMessage());
		}
		return client;
	}

	/**
	 * 探索ノードを切り離す
	 */
	public void unbindSearchNode(String robotId) throws RemoteException {
		log.info("探索ノード'" + robotId + "'を切り離しています");

		boolean exitf = true;
		if(detachSearchNode(robotId)) {
			exitf = false;
		}

		if(exitf) {
			// 誰も管理していない探索ノードだった
			log.info("管理されていない探索ノード'" + robotId + "'を切り離そうとしました");
			throw new RemoteException("no routing manager manages the robot '" + robotId + "'.");
		}

		log.info("探索ノード'" + robotId + "'が切り離されました");
	}

	/**
	 * 探索ノードをバインドする
	 */
	public RoutingNodeClient attachSearchNode() throws RemoteException {
		RoutingNodeClientImpl rn = new RoutingNodeClientImpl(this);
		searchNodes.put(rn.getNodeId(), rn);

		// 分配アルゴリズムにノードを追加する
		allocater.addChildNode(rn.getNodeId());

		return rn;
	}

	/**
	 * 探索ノードを切り離す
	 * @param robotId 探索ノードのID
	 */
	public boolean detachSearchNode(String robotId) {
		boolean rc = false;
		synchronized(searchNodes) {
			RoutingNodeClient rnc = (RoutingNodeClient)searchNodes.get(robotId);
			if(rnc == null || searchNodes.remove(robotId) == null) {
				// 探索ノードの切り離しに失敗した場合
				// 通常発生しないが，既に切り離されていた場合など
				rc = false;
			} else {
				// 分配アルゴリズムからノードを削除する
				allocater.removeChildNode(robotId);
				// 探索ノードにリースしているリモートオブジェクトを回収する
				try {
					UnicastRemoteObject.unexportObject(rnc, true);
				} catch (NoSuchObjectException e) {
					// 無視
				}
				rc = true;
			}
		}

		return rc;
	}

	/**
	 * 未処理URLをファクトリのプールから削除する
	 */
	public void removeUnTracedURL(String url) throws RemoteException {
		try {
			synchronized(unTracedURLs) {
				unTracedURLs.remove(url);
			}
		} catch(MalformedURLException ue) {
			throw new RemoteException(ue.getMessage());
		}
	}

	/**
	 * ファクトリに新しいURLを追加する
	 */
	public void addUnTracedURL(String url) throws RemoteException {
		try {
			unTracedURLs.add(url);
		} catch(MalformedURLException ue) {
			throw new RemoteException(ue.getMessage());
		}
	}

	/**
	 * 適当な下位ノードに対して，未処理URLを割り当てる
	 */
	public void allocateURL() {
		try {
			synchronized(unTracedURLs) {
				String[] domains = unTracedURLs.domains();
				for(int i = 0; i < domains.length; i++) {
					String domain = domains[i];
					String[] urls = unTracedURLs.urls(domain);

					for(int j = 0; j < urls.length; j++) {
						try {
							String destNodeId = allocater.allocate(urls[j]);
							rejectedCommandQueue.add(new SyncCommand(urls[j], SyncCommand.COMMAND_FORWARD_URL, getNodeId(), destNodeId));
							unTracedURLs.remove(urls[j]);
						} catch(RemoteException re) {
							re.printStackTrace();
						} catch(AllocationException ae) {
							ae.printStackTrace();
						} catch(SyncQueueException se) {
							// 捨てられた
						} catch(MalformedURLException ue) {
							// 捨てられた
						}
					}
				}
			}
		} catch(NoSuchElementException nse) {
			// リストが空になった場合．何もしない
		}
	}

	/**
	 * 探索の許可されているIPアドレス範囲の一覧を返す．
	 * @return 探索が許可されているIPアドレス範囲のリスト
	 * @throws RemoteException 処理中に発生した例外
	 */
	public String[] getAllowedRoutingIPRanges() throws RemoteException {
		return config.getRoutingIPRanges();
	}

	/**
	 * 探索ノードの一覧を返す
	 * @return 探索ノード名
	 * @throws RemoteException 処理中に発生した例外
	 */
	public String[] getClientNames() throws RemoteException {
		Vector<String> tmp = new Vector<String>();
		synchronized(searchNodes) {
			Enumeration e = searchNodes.keys();
			while(e.hasMoreElements()) {
				tmp.addElement(e.nextElement().toString());
			}
		}

		if(tmp.size() == 0) {
			return null;
		} else {
			return (String[])tmp.toArray(new String[1]);
		}
	}

	/**
	 * 指定されたノードに宛てた探索ロックの差分情報を返す
	 * @param rk ノードID
	 * @return 自分宛てに届いた同期コマンド
	 * @throws RemoteException 処理中に発生した例外
	 */
	public SyncCommand[] getModifiedSyncData(String rk) throws RemoteException {
		allocateURL();				// URLプールに残っているものを，コマンドへ変換してみる
		SyncCommand[] o = rejectedCommandQueue.get(rk);
		rejectedCommandQueue.clear(rk);
		return o;
	}

	/**
	 * 探索ロックの差分情報を保存する．
	 * @param locks 同期コマンド
	 * @throws Exception 同期コマンド処理中に発生した例外
	 */
	public void sync(SyncCommand[] locks) throws Exception {
		if(locks == null || locks.length == 0) {
			return;
		}
		log.info(locks.length + "件の同期データを受信しました");

		Vector<SyncCommand> receivedCommands = new Vector<SyncCommand>(Arrays.asList(locks));
		SyncCommand[] rejectedCommands = syncQueue.add(locks);
		receivedCommands.removeAll(Arrays.asList(rejectedCommands));	// 受信したコマンド - 拒否されたもの = 受理されたもの

		for(int i = 0; i < receivedCommands.size(); i++) {
			SyncCommand command = (SyncCommand)receivedCommands.elementAt(i);
			if(command.getType() == SyncCommand.COMMAND_FORWARD_URL) {
				// 未処理URLを受信した場合，URLプールに格納する
				unTracedURLs.add((String)command.getObject());
			} else if(command.getType() == SyncCommand.COMMAND_SHUTDOWN) {
				// シャットダウンコマンドの場合，そのまま受理する
				rejectedCommandQueue.add(command);
			}
		}

		// out queueに拒否されたコマンドを登録する
		rejectedCommandQueue.add(rejectedCommands);
	}

	public static void main(String[] args) {
		try {
			Argument[] arguments = ArgumentParser.parse(args);
			String filename = null;
			for(int i = 0; i < arguments.length; i++) {
				Argument arg = arguments[i];
				if(arg.isOption()) {
					if("config".equals(arg.getName())) {
						filename = arg.getValue();
					}
				}
			}
			RNodeClientFactoryImpl factory = null;
			if(filename == null) {
				factory = new RNodeClientFactoryImpl();
			} else {
				factory = new RNodeClientFactoryImpl(filename); 
			}
			factory.startModule();
		} catch(Exception e) {
			System.err.println(e.getMessage());
			e.printStackTrace();
		}
	}

	/**
	 * モジュールコントロール用インターフェース
	 * モジュールを開始する
	 */
	public void startModule() throws ModuleException {
		try {
			RSSRoutingNodeConfig conf = new RSSRoutingNodeConfig(rnodeConfigFilename);

			init(conf.getRSSDSR());

			// 分配アルゴリズムの設定
			allocater = conf.getAllocatingAlgorithm();
			allocater.setRoutingAlgorithm(conf.getRoutingAlgorithm());

			// configに設定されている初期URLを，未処理のものとして登録する
			String defaultURLsLine = conf.getRSSFactoryDefaultURLs();
			if(defaultURLsLine != null) {
				String[] defaultURLs = defaultURLsLine.split(",");
				for(int i = 0; i < defaultURLs.length; i++) {
					addUnTracedURL(defaultURLs[i]);
				}
			}
			config = conf;

			log.info("IPアドレス範囲設定: 全" + conf.getRoutingIPRanges().length + "件");
			log.info("RNodeClientFactory '" + factoryServiceName + "' is started");
			log.info("経路決定ノードクライアントファクトリサービス'" + factoryServiceName + "'が開始されました");
		} catch(Exception e) {
			throw new ModuleException(e);
		}
	}

	/**
	 * モジュールコントロール用インターフェース::
	 * モジュールを一時停止する
	 */
	public void suspendModule() throws ModuleException {
	}

	/**
	 * モジュールコントロール用インターフェース::
	 * 停止しているモジュールの実行を再開する
	 */
	public void continueModule() throws ModuleException {
	}

	/**
	 * モジュールコントロール用インターフェース::
	 * モジュールを停止する
 	 * このメソッドが呼ばれた場合，モジュールは安全な方法で終了する
	 * 終了したモジュールは，新しくインスタンスが生成されないと開始することはできない
	 */
	public void stopModule() throws ModuleException {
	}

	/**
	 * サービス名を返す
	 * @return サービス名
	 * @throws RemoteException サービス名取得に失敗
	 */
	public String getName() throws RemoteException {
		return factoryServiceName;
	}
	/**
	 * 完全修飾サービス名を返す
	 * @return 完全修飾サービス名
	 * @throws RemoteException サービス名取得に失敗
	 */
	public String getFqsn() throws RemoteException {
		return factoryServiceName + "@" + config.getRSSDSR();
	}
	/**
	 * サービスを停止する
	 * RMIサーバーとして実装されている場合，unexportされる
	 * @param force true:強制停止 / false:安全に停止
	 * @throws RemoteException サービス停止に失敗
	 */
	public void shutdown(boolean force) throws RemoteException {
		log.info("サーバーを停止します");

		// 全探索ノードに対して，停止コマンドを発行する
		String[] clientNodes = getClientNames();
		if(clientNodes != null && clientNodes.length > 0) {
			log.info("探索ノードの停止を待っています..");
			SyncCommand[] commands = new SyncCommand[clientNodes.length];
			for(int i = 0; i < clientNodes.length; i++) {
				SyncCommand cmd = new SyncCommand();
				cmd.setType(SyncCommand.COMMAND_SHUTDOWN);
				cmd.setTo(clientNodes[i]);
				cmd.setFrom(getNodeId());
				cmd.setObject(clientNodes[i]);
				commands[i] = cmd;
			}
			try {
				sync(commands);
			} catch(RemoteException re) {
				throw re;
			} catch (Exception e) {
				e.printStackTrace();
				throw new RemoteException(e.getMessage());
			}

			log.info("探索ノードが全て停止しました");
		}

		// 同期コマンドが成功した場合，探索ノードが全て停止するのを待つ
		try {
			while(searchNodes.size() > 0) {
				Thread.sleep(1000);
			}
		} catch (InterruptedException e) {
			log.warn("探索ノード停止待ち中に，予期しない割り込みが発生しました：", e);
		}

		// DSRから切り離す
		try {
			log.info("経路決定ノードをDSRから切り離しています");
			LookupFactory.getLookup(DSRURL).unbind(RSS_SERVID_ROUTING_NODE_CLIENT_FACTORY, factoryServiceName);
		} catch (Exception e1) {
			throw new RemoteException(e1.getMessage());
		}

		UnicastRemoteObject.unexportObject(this, force);

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