/*
 * $Id: RSSDistributedServiceRegistryImpl.java,v 1.25 2005/10/16 15:44:54 rampil Exp $
 * Copyright (c) 2005 LOGICAL-PARADOX.ORG
 */
package org.logical_paradox.rss.dsr;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.rss.ModuleException;
import org.logical_paradox.rss.RSSSubModule;
import org.logical_paradox.rss.Service;
import org.logical_paradox.rss.dsr.algorithm.ServiceBindingAlgorithm;
import org.logical_paradox.rss.dsr.routing.Node;
import org.logical_paradox.rss.lookup.LookupFactory;
import org.logical_paradox.rss.util.RandomKeyGenerator;

/**
 * RSSDistributedServiceRegistryImpl
 * 分散サービスレジストリ(実装)
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.25 $
 */
public class RSSDistributedServiceRegistryImpl extends UnicastRemoteObject implements RSSDistributedServiceRegistry, RSSSubModule {
	/** ロガー */
	private static final Log logger = LogFactory.getLog(RSSDistributedServiceRegistryImpl.class);

	/** コンフィグファイルのデフォルトファイル名 */
	public static final String RSS_DSRCONF_PATH = "conf/rssdsr.conf";

	/** コンフィグレーション */
	private RSSDSRConfig cfg = null;
	/** このDSRに登録されているリモートサービス */
	private final Hashtable remoteServices;
	/** サービスバインディングアルゴリズム */
	private final Hashtable bindingAlgorithms;
	/** 接続先URI(親DSR) */
	private String endPointURI = null;
	/** コンフィグパス */
	private String dsrConfigPath = null;
	/** このノード */
	private Node node;

	/** オーバーレイネットワーク監視スレッド */
	private NetworkKeepAgent monitor;

	/** このDSRのサービス名 */
	private String dsrServiceName;

	/**
	 * コンストラクタ
	 * @throws RemoteException インスタンス化に失敗した
	 */
	public RSSDistributedServiceRegistryImpl() throws RemoteException {
		remoteServices = new Hashtable();
		bindingAlgorithms = new Hashtable();
	}
	/**
	 * コンストラクタ
	 * @param endp 接続先DSR
	 * @throws RemoteException DSRの生成に失敗
	 */
	public RSSDistributedServiceRegistryImpl(String endp) throws RemoteException {
		this();
		endPointURI = endp;
	}
	/**
	 * サービス名を一覧化する
	 * @param type 種類
	 * @return 登録されているサービス名の配列
	 */
	public String[] services(int type) throws RemoteException {
		ServiceList slist = (ServiceList)remoteServices.get(new Integer(type));
		if(slist == null) {
			return new String[0];
		}
		return slist.serviceNames();
	}
	/**
	 * 指定ノードに対してpingを打つ
	 * @return 相手先ノードがメッセージを受信した時間(time in millis)
	 * @throws RemoteException pingが到達しなかった
	 */
	public long ping() throws RemoteException {
		return DSRConstant.datatime.currentTimeInMillis();
	}
	/**
	 * コンフィグを返す
	 * @return コンフィグレーション
	 * @throws RemoteException コンフィグの取得に失敗
	 */
	public RSSDSRConfig getConfig() throws RemoteException {
		return cfg;
	}
	/**
	 * DSR間のリンクを再構築する
	 * @throws RemoteException 再構築処理に失敗
	 */
	public void configure() throws RemoteException {
		logger.info("DSR間のリンクを再構築しています");
		synchronized(node.getRegistry()) {
			node.configure();
		}
		logger.info("DSR間のリンクを再構築しました");
	}
	/**
	 * ノードIDを指定して,リモートサービスをバインドする
	 * @param type タイプ
	 * @param serviceName ノードID
	 * @param stub 登録するリモートサービス
	 * @return 生成されたID(指定したノードIDと同じものが返る
	 * @throws RemoteException 登録に失敗した際に発生する例外
	 */
	public String bind(int type, String serviceName, Service stub) throws RemoteException {
		ServiceList slist = (ServiceList)remoteServices.get(new Integer(type));

		// まだサービスが何も登録されていなかった場合
		// ServiceListオブジェクトを作成する
		if(slist == null) {
			slist = new ServiceList(getBindingAlgorithmByType(type));
		}

		slist.add(serviceName, stub);
		remoteServices.put(new Integer(type), slist);

		logger.info("リモートサービスを追加しました: " + serviceName + "(" + type + ")");

		return serviceName;
	}
	/**
	 * リモートサービスを登録する
	 * @param type サービスタイプ
	 * @param stub 登録するリモートサービス
	 * @return 生成されたID
	 * @throws RemoteException 登録に失敗した際に発生する例外
	 */
	public String bind(int type, Service stub) throws RemoteException {
		return bind(type, getRemoteServiceName(type), stub);
	}
	/**
	 * 指定されたサービスの登録を解除する
	 * @param type サービスタイプ
	 * @param nodeId 切り離すノードID
	 * @throws RemoteException 登録解除に失敗した際に発生する例外
	 */
	public void unbind(int type, String nodeId) throws RemoteException {
		ServiceList slist = (ServiceList)remoteServices.get(new Integer(type));

		if(slist == null || slist.contains(nodeId) == false) {
			// 指定されたサービスが登録されていない場合
			throw new RemoteException("タイプ" + type + "に対応するリモートサービスは，登録されていません");
		}

		slist.remove(nodeId);

		logger.info("リモートサービスを切り離しました: " + nodeId + "(" + type + ")");
	}
	/**
	 * 指定されたサービス名のスタブを返す
	 * 指定されたトレースレベルまで探索する<br>
	 * トレースレベル1は、ローカルのみを対象として検索することを意味する
	 * @param name 検索するサービス名
	 * @param traceLevel トレース深度(1=ローカルのみ)
	 * @return 検索にヒットしたサービスのスタブ
	 * @throws RemoteException 検索に失敗した
	 */
	public Service getService(String name, int traceLevel) throws RemoteException {
		// 与えられた検索対象サービス名が，まさに自分だった場合は自分を返す
		if(getName().equals(name)) {
			return this;
		}
		// 適当な深度まで検索する
		Message msg = new Message();
		msg.setTargetName(name);
		msg.setTraceLevel(traceLevel);

		// 検索
		Service[] services = discover(msg);

		if(services == null || services.length == 0) {
			// 何も発見できなかった
			return null;
		} else {
			// あった
			return services[0];
		}
	}
	/**
	 * 指定されたサービス名のスタブを返す
	 * @param name 検索するサービス名
	 * @return 検索にヒットしたサービスのスタブ
	 * @throws RemoteException 検索に失敗した
	 */
	public Service getService(String name) throws RemoteException {
		return getService(name, getConfig().getTraceLevel());
	}
	/**
	 * 指定されたサービスに対応するスタブを返す．
	 * @param type 検索するサービスタイプ
	 * @param traceLevel 探索深度
	 * @return 検索して発見されたサービス
	 * @throws RemoteException 検索中に発生した障害
	 */
	public Service getService(int type, int traceLevel) throws RemoteException {
		Message msg = new Message();
		msg.setTargetName(null);
		msg.setDiscoverType(type);
		msg.setTraceLevel(traceLevel);

		// 検索
		Service[] services = discover(msg);

		if(services == null || services.length == 0) {
			return null;
		} else {
			return services[0];
		}
	}
	/**
	 * 指定されたサービスに対応するスタブを返す．
	 * 探索深度はコンフィグレーションの値を使用する．
	 * @param type サービスタイプ
	 * @return スタブ
	 * @throws RemoteException サービスの取得に失敗
	 */
	public Service getService(int type) throws RemoteException {
		// ローカルのみ検索
		return getService(type, getConfig().getTraceLevel());
	}
	/**
	 * サービス名を作成する
	 * @param type サービスタイプ
	 * @return 自動的に生成されたサービス名(ノードID)
	 * @throws IllegalArgumentException サービスタイプが不正
	 */
	protected String getRemoteServiceName(int type) throws IllegalArgumentException {
		String theKindOfService = null;

		switch(type) {
			// DSR - DSRのサービス名は自動生成できない
			case RSS_SERVID_SERVICE_LOCATER:
				throw new IllegalArgumentException("dsr service name cannot be generated automatically.");
			// 経路決定ノード
			case RSS_SERVID_ROUTING_NODE_CLIENT_FACTORY:
				theKindOfService = "RNCF/";
				break;
			// コンテンツ管理システム
			case RSS_SERVID_REMOTE_CONTENTS_MGR:
				theKindOfService = "RCMGR/";
				break;
			// インデックスシステム
			case RSS_SERVID_INDEX_SERVER:
				theKindOfService = "IDXSRV/";
				break;
			// キューシステム
			case RSS_SERVID_OQS4R:
				theKindOfService = "OQS4R/";
				break;

			default:
				throw new IllegalArgumentException("SEVICE ID:" + type + "は，定義されていません．");
		}

		String serviceName = theKindOfService + RandomKeyGenerator.getUniqKey(8);

		return serviceName;
	}

	/**
	 * リモートサービス選択用のアルゴリズムを返す
	 * @param type サービスタイプ
	 * @return サービス選択アルゴリズム
	 * @throws IllegalArgumentException 未定義のサービスタイプが指定された
	 */
	protected ServiceBindingAlgorithm getBindingAlgorithmByType(int type) throws IllegalArgumentException {
		switch(type) {
			case RSS_SERVID_SERVICE_LOCATER:
				return cfg.getDSRBindingAlgorithm();

			case RSS_SERVID_ROUTING_NODE_CLIENT_FACTORY:
				return cfg.getRNodeBindingAlgorithm();

			case RSS_SERVID_REMOTE_CONTENTS_MGR:
				return cfg.getRCMGRBindingAlgorithm();

			case RSS_SERVID_INDEX_SERVER:
				return cfg.getIndexServerBindingAlgorithm();

			case RSS_SERVID_OQS4R:
				return cfg.getOqs4rBindingAlgorithm();

			default:
				throw new IllegalArgumentException("SEVICE ID:" + type + "は，定義されていません．");
		}
	}
	/**
	 * 接続先DSRを設定する
	 * @param uri URI
	 */
	protected void setEndPointURI(String uri) {
		endPointURI = uri;
	}
	/**
	 * dsr-configのパスを指定する
	 * @param path パス
	 */
	protected void setDsrConfigPath(String path) {
		dsrConfigPath = path;
	}
	/**
	 * レジストリにインスタンスをバインドする
	 * RMIを使用したくない場合，このメソッドをオーバーライドする
	 * @param dsr バインドするサーバーインスタンス
	 * @throws RemoteException バインドに失敗
	 * @throws MalformedURLException バインド先がおかしい
	 */
	protected void bindToRegistry(RSSDistributedServiceRegistry dsr) throws RemoteException, Exception {
		logger.info("DSR[" + dsrServiceName + "]を登録しています");
		LookupFactory.register(this);
	}
	/**
	 * モジュールコントロール用インターフェース::
	 * モジュールを開始する
	 */
	public void startModule() throws ModuleException {
		try {
			if(dsrConfigPath == null) {
				cfg = new RSSDSRConfig(RSSDistributedServiceRegistryImpl.RSS_DSRCONF_PATH);
			} else {
				cfg = new RSSDSRConfig(dsrConfigPath);
			}

			dsrServiceName = cfg.getRSSDistributedServiceRegistry();

			node = new Node(this, dsrServiceName);

			// ネットワーク監視スレッドの開始
			monitor = new NetworkKeepAgent(node);

			bindToRegistry(this);

			// 自分自身をバインドする
			bind(RSS_SERVID_SERVICE_LOCATER, dsrServiceName, this);

			// DSRの接続先が指定されている場合,他のDSRと相互接続する
			if(endPointURI != null && endPointURI.trim().length() > 0) {
				RSSDistributedServiceRegistry buddy = (RSSDistributedServiceRegistry)LookupFactory.getLookup(endPointURI).hug(getName(), this);
				// バディを自分のレジストリに登録する
				// バディのサービス名は自動生成しない(lookupできなくなるので)
				String buddyName = bind(RSS_SERVID_SERVICE_LOCATER, endPointURI, buddy);
				logger.info("DSR[" + buddyName + "]と接続しました");

				// 自分のノードに対して、接続相手を追加する
				node.add(new Node(buddy, buddyName));
			}

			logger.info("分散サービスレジストリ '" + dsrServiceName + "' が開始されました．");

			// オーバーレイネットワーク監視スレッドの開始
			monitor.start();
			logger.info("オーバーレイネットワーク監視スレッドを開始しました");

		} catch(Exception e) {
			logger.error("モジュール開始処理中に継続不能な例外が発生しました:", e);
			throw new ModuleException(e);
		}
	}

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

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

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

	/**
	 * メインメソッド
	 * @param args プログラム引数
	 * @throws Exception 処理されなかった例外
	 */
	public static void main(String[] args) throws Exception {
		try {
			RSSDistributedServiceRegistryImpl selector = new RSSDistributedServiceRegistryImpl();
			System.err.println("引数=" + args.length);
			if(args.length >= 2) {
				// 他のDSRに接続する
				selector.setEndPointURI(args[1]);
			}
			if(args.length >= 3) {
				selector.setDsrConfigPath(args[2]);
			}
			selector.startModule();
		} catch(Exception e) {
			System.err.println(e.getMessage());
			e.printStackTrace();
		}
	}
	/**
	 * 指定されたタイプのサービスを分散ネットワークから検索し,スタブを返す
	 * @param msg 探索条件
	 * @return 指定探索範囲で発見されたサービスのスタブ
	 */
	public Service[] discover(Message msg) throws RemoteException {
		// 自分のレジストリを調べる
		ArrayList discoveredServices = new ArrayList();

		ServiceList services = null;
		if(msg.getDiscoverType() > 0) {
			// サービスタイプが指定されている場合
			services = (ServiceList)remoteServices.get(new Integer(msg.getDiscoverType()));
		} else {
			// サービスタイプが指定されていない場合は名前で探す
			for(Iterator sit = remoteServices.values().iterator(); sit.hasNext();) {
				ServiceList sl = (ServiceList)sit.next();
				if(sl.contains(msg.getTargetName())) {
					return new Service[]{ sl.getService(msg.getTargetName()) };
				}
			}
		}

		if(msg.getTraceLevel() == 1) {
			// サービス探索深度が1の場合，ここで終了
			return (Service[])discoveredServices.toArray(new Service[0]);
		}
		if(services != null) {
			discoveredServices.addAll(services.values());
		}

		// 自分のレジストリに存在するDSRに対して、発見要求を送信する
		// 接続しているDSRがあり、かつ探索深度が既定以下であれば､他のDSRを検索する
		ServiceList registeredDsr = (ServiceList)remoteServices.get(new Integer(RSSDistributedServiceRegistry.RSS_SERVID_SERVICE_LOCATER));
		if(registeredDsr != null && msg.getCurrentLevel() < cfg.getTraceLevel()) {
			logger.info("発見スレッドを生成します");
			// いくつかのDSRが存在する場合，並列で発見要求を送信する
			Service[] dsrs = (Service[])registeredDsr.values().toArray(new Service[0]);
			DiscoveryThread[] threads = new DiscoveryThread[dsrs.length];

			for(int i = 0; i < dsrs.length; i++) {
				// マスターのメッセージをコピーして、新しいインスタンスを生成する．
				Message param = new Message(msg);
				// 現在の探索深度 +1
				param.incrementLevel();
				// 現在のノードを探索履歴に追加する
				threads[i] = new DiscoveryThread((RSSDistributedServiceRegistry)dsrs[i], param);
				threads[i].start();
				logger.info("発見スレッド: " + dsrs[i].getName() + " → 開始");
			}
			// スレッドの終了待ち合わせ
			try {
				for(int i = 0; i < dsrs.length; i++) {
					threads[i].join();
				}
			} catch (InterruptedException e) {
				// 発見スレッドが何らかの原因で停止した
				logger.info("発見スレッドが停止しました: 【原因】" + e.getMessage());
			}
			logger.info("発見スレッドが全て終了");
			// 探索結果の収集
			for(int i = 0; i < threads.length; i++) {
				discoveredServices.addAll(Arrays.asList(threads[i].getResult()));
			}
		}

		return discoveredServices.size() == 0 ? new Service[0] : (Service[])discoveredServices.toArray(new Service[0]);
	}
	/**
	 * 指定名称のサービスに到達するまでの経路をトレースし、経路情報を返す
	 * @param name サービス名
	 * @return トレース情報
	 * @throws RemoteException トレースに失敗した際に発生する例外
	 */
	public Message trace(String name) throws RemoteException {
		// TODO 自動生成されたメソッド・スタブ
		return null;
	}
	/**
	 * サービス名を返す
	 * @return サービス名
	 * @throws RemoteException サービス名取得に失敗
	 */
	public String getName() throws RemoteException {
		return dsrServiceName;
	}
	/**
	 * 完全修飾サービス名を返す
	 * @return 完全修飾サービス名
	 * @throws RemoteException サービス名取得に失敗
	 */
	public String getFqsn() throws RemoteException {
		return dsrServiceName;
	}
	/**
	 * サービスを停止する
	 * RMIサーバーとして実装されている場合，unexportされる
	 * @param force true:強制停止 / false:安全に停止
	 * @throws RemoteException サービス停止に失敗
	 */
	public void shutdown(boolean force) throws RemoteException {
		logger.info("サーバーを停止します");

		logger.info("オーバーレイネットワーク維持管理スレッドを停止します．");
		if(monitor != null) {
			monitor.shutdown();
			try {
				monitor.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		// DSR本体を停止する
		UnicastRemoteObject.unexportObject(this, force);

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

	/**
	 * DSRが構成するオーバーレイネットワークを維持管理するスレッド]
	 * 一定時間ごとに接続状態を確認し、ネットワークを最適の状態に維持する
	 * @author satoshi akabane@logical-paradox.org
	 * @version $Revision: 1.25 $
	 */
	class NetworkKeepAgent extends Thread {
		/** ロガー */
		private final Log log = LogFactory.getLog(NetworkKeepAgent.class);
		/** 監視対象ノード */
		private final Node node;
		/** 監視処理中かどうか(true:監視処理中 / false:監視停止中) */
		private boolean running = false;

		/**
		 * コンストラクタ
		 * @param n 監視対象ノード
		 */
		protected NetworkKeepAgent(Node n) {
			node = n;
			setDaemon(true);
		}
		/**
		 * スレッドメイン
		 */
		public void run() {
			running = true;
			try {
				while(running) {
					// DSR間のリンクを再構築する
					node.getRegistry().configure();

					Thread.sleep(node.getRegistry().getConfig().getMonitorInterval());
				}
			} catch(InterruptedException e) {
				// 監視スレッドの停止
				log.info("何らかの理由により、DSR監視スレッドを停止します:[" + e.getMessage() + "]");
				running = false;
			} catch (RemoteException e) {
				// コンフィグの取得に失敗
				log.error("ネットワーク上の問題により、DSR監視スレッドを停止します:[" + e.getMessage() + "]");
				running = false;
			}
		}
		/**
		 * スレッドを停止する．
		 */
		public void shutdown() {
			interrupt();
		}
	}
	/**
	 * 他のDSRに対して発見要求を送信し、結果を受信するスレッド
	 * 他のDSRに対する要求は、並列で実行される
	 * 
	 * @author satoshi akabane@logical-paradox.org
	 * @version $Revision: 1.25 $
	 */
	class DiscoveryThread extends Thread {
		/** 他DSR */
		private RSSDistributedServiceRegistry dsr;
		/** 検索結果 */
		private Service[] discoveryResult;
		/** 問い合わせ条件 */
		private Message msg;

		/**
		 * コンストラクタ
		 * @param d 他DSR
		 * @param m 問い合わせメッセージ
		 */
		protected DiscoveryThread(RSSDistributedServiceRegistry d, Message m) {
			dsr = d;
			msg = m;
		}
		/**
		 * スレッドを実行する
		 */
		public void run() {
			try {
				discoveryResult = dsr.discover(msg);
			} catch (RemoteException e) {
				// 通信エラーなど
				e.printStackTrace();
				discoveryResult = new Service[0];
			}
		}
		/**
		 * 検索結果を取り出す
		 * @return 検索結果
		 */
		public Service[] getResult() {
			return discoveryResult == null ? new Service[0] : discoveryResult;
		}
	}
}

// end of RSSDistributedServiceRegistryImpl.java
