/*
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 * $Id: DomainFileSystemImpl.java,v 1.2 2005/09/07 13:38:06 rampil Exp $
 */
package org.logical_paradox.vsan.dfs;

import java.rmi.Naming;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.ResourceBundle;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.vsan.dfs.event.DomainUpdateEventObject;
import org.logical_paradox.vsan.dfs.event.DomainUpdateListener;
import org.logical_paradox.vsan.domain.Domain;
import org.logical_paradox.vsan.domain.DomainController;
import org.logical_paradox.vsan.lookup.LookupFactory;

/**
 * ドメインファイルシステムの実装
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.2 $
 */
public class DomainFileSystemImpl extends UnicastRemoteObject implements DomainFileSystem, DomainUpdateListener {
	/** ロガー */
	private static final Log log = LogFactory.getLog(DomainFileSystemImpl.class);

	/** このドメインファイルシステムのコンフィグレーション */
	private final DfsConfig config;
	/** ノードID */
	private final String nodeId;

	/** ドメインコントローラー */
	private DomainController domainController;

	/** ドメイン */
	private Domain domain;

	/** ドメイン情報自動更新スレッド */
	private DomainSynchroThread domainSynchronizer;

	/** クライアント接続モニタースレッド */
	private ClientConnectionMonitor clientMonitor;

	/** ドメイン接続プロバイダー */
	private DomainConnectionProvider domainConnectionProvider;

	/** ドメイン接続プロキシ */
	private final ArrayList proxies = new ArrayList();

	/**
	 * コンストラクタ
	 * @param config コンフィグ
	 * @throws RemoteException インスタンス化に失敗
	 */
	public DomainFileSystemImpl(DfsConfig c) throws RemoteException {
		config = c;
		// 今のところ、ノードID = 接続しているドメイン名とする
		nodeId = config.getDomainPreconnectId();

		// 初期化処理
		init();
	}

	/**
	 * 初期接続を確立する
	 */
	protected void init() throws RemoteException {
		log.info("ドメインファイルシステムを初期化しています");

		// ドメイン情報を取得する
		domainController = LookupFactory.getLookup().lookupDomainController(config.getDomainControllerId());
		domain = domainController.getDomain(config.getDomainPreconnectId());

		// 接続プロバイダーを作成する
		log.info("ドメイン接続プロバイダーを作成してます");
		domainConnectionProvider = new DomainConnectionProvider(config, domain);

		if(config.getDomainSynchroInterval() > 0) {
			domainSynchronizer = new DomainSynchroThread(config, domainController, domain);
			domainSynchronizer.addListener(this);
			domainSynchronizer.start();
			log.info("ドメイン情報自動更新スレッドを作成しました");
		}

		if(config.getClientMonitorInterval() > 0) {
			clientMonitor = new ClientConnectionMonitor(config.getClientMonitorInterval());
			clientMonitor.start();
			log.info("クライアント接続モニターを作成しました");
		}
		log.info("ドメインファイルシステムを初期化しました");
	}
	/**
	 * ドメイン変更通知イベント．
	 * @param e ドメイン変更オブジェクト
	 */
	public void domainChanged(DomainUpdateEventObject e) {
		log.info("ドメイン情報に変更が発生したため、ノード情報を最新化します");
		ArrayList updateInfo = new ArrayList();
		List before = new ArrayList(e.getDomainBefore().getNodeList().keySet());
		List after = new ArrayList(e.getDomainAfter().getNodeList().keySet());

		// ドメインから離脱したノードを調べる
		for(Iterator it = before.iterator(); it.hasNext();) {
			String node = (String)it.next();
			if(after.contains(node) == false) {
				// 以前あったのに今はない = 離脱したノード
				log.info("ノード[" + node + "]が離脱しました");
				updateInfo.add(new DomainUpdateDetail(DomainUpdateDetail.UPDATE_TYPE_LEAVE, node));			
			}
		}
		// 新しくドメインに参加したノードを調べる
		for(Iterator it = after.iterator(); it.hasNext();) {
			String node = (String)it.next();
			if(before.contains(node) == false) {
				// 以前なかったのに今はある = 追加されたノード
				log.info("ノード[" + node + "]が参加しました");
				updateInfo.add(new DomainUpdateDetail(DomainUpdateDetail.UPDATE_TYPE_JOIN, node));
			}
		}

		// 今回の更新を，このドメインファイルシステムに接続している
		// クライアントに対して通知する
		notifyChange(updateInfo);

		log.info("ドメイン情報を更新しました");
	}
	/**
	 * ドメイン更新情報を，ドメインファイルシステムに接続しているクライアントに通知する．
	 * @param changes 更新内容のリスト
	 */
	protected void notifyChange(List changes) {
		log.info("ドメイン情報の更新内容について、クライアントに通知します");
		DomainConnectionProxyImpl[] localProxies = (DomainConnectionProxyImpl[])proxies.toArray(new DomainConnectionProxyImpl[0]);

		for(Iterator it = changes.iterator(); it.hasNext();) {
			DomainUpdateDetail dud = (DomainUpdateDetail)it.next();

			if(dud.getType() == DomainUpdateDetail.UPDATE_TYPE_JOIN) {
				// 追加
				for(int i = 0; i < localProxies.length; i++) {
					localProxies[i].getConnection().join(dud.getNode());
				}
			} else if(dud.getType() == DomainUpdateDetail.UPDATE_TYPE_LEAVE) {
				// 削除
				for(int i = 0; i < localProxies.length; i++) {
					localProxies[i].getConnection().leave(dud.getNode());
				}
			}
		}
		log.info("ドメイン情報の更新内容について、クライアントに通知しました");
	}
	/**
	 * ドメインファイルシステムに対してアクセスするための接続オブジェクトを返す
	 * 接続数の状態によっては、取得できない場合もある
	 * @param clientNodeId クライアントノードのID
	 * @return 取得できたドメイン接続オブジェクト
	 * @throws RemoteException 取得中にネットワーク障害などが発生した
	 */
	public DomainConnectionProxy getConnection(String clientNodeId) throws RemoteException {
		// このクライアントに割り当てられたドメイン接続オブジェクトを返す
		// 接続数によっては、割り当てができない場合もある
		DomainConnection connection = domainConnectionProvider.getConnection(clientNodeId);
		DomainConnectionProxy proxy = new DomainConnectionProxyImpl(connection, clientNodeId, config);

		proxies.add(proxy);

		return proxy;
	}
	/**
	 * このドメインファイルシステムのノードIDを返す
	 * @return サーバーインスタンス生成時に割り当てられたノードID
	 * @throws RemoteException 通信障害など
	 */
	public String getNodeId() throws RemoteException {
		return nodeId;
	}

	/**
	 * メインメソッド
	 * @param args プログラム引数
	 * @throws Exception 処理されなかった例外
	 */
	public static final void main(String[] args) throws Exception {
		log.info("ドメインファイルシステムを開始しています..");

		/** リソースの読み込み */
		ResourceBundle resource = ResourceBundle.getBundle(DfsConstant.RESOURCE_PREFIX);
		DfsConfig config = new DfsConfig(resource);

		log.info("サーバーコンフィグレーションの読み込み終了.");

		// サーバーインスタンスの作成
		DomainFileSystem filesystem = new DomainFileSystemImpl(config);

		// RMIレジストリにバインド
		Naming.rebind(filesystem.getNodeId(), filesystem);

		log.info("ドメインファイルシステム[" + filesystem.getNodeId() + "]を開始しました.");
	}

	/**
	 * ドメイン更新情報
	 * @author satoshi akabane@logical-paradox.org
	 * @version $Revision: 1.2 $
	 */
	class DomainUpdateDetail {
		/** 更新区分 - 新規参加 */
		public static final int UPDATE_TYPE_JOIN = 1;
		/** 更新区分 - 離脱 */
		public static final int UPDATE_TYPE_LEAVE = 2;

		/** 更新区分 */
		private final int type;
		/** 対象ノード */
		private String node;

		/**
		 * コンストラクタ
		 * @param type_ 更新区分
		 * @param node_ ノード
		 */
		public DomainUpdateDetail(int type_, String node_) {
			type = type_;
			node = node_;
		}
		/**
		 * 更新区分を返す．
		 * @return 更新区分
		 */
		public int getType() {
			return type;
		}
		/**
		 * ノードを返す．
		 * @return ノード
		 */
		public String getNode() {
			return node;
		}
	}
	/**
	 * ドメインコントローラーに定期的にアクセスし、ノード情報を更新するスレッド
	 * @author satoshi akabane@logical-paradox.org
	 * @version $Revision: 1.2 $
	 */
	class DomainSynchroThread extends Thread {
		/** ロガー */
		private final Log log = LogFactory.getLog(DomainSynchroThread.class);

		/** ドメインファイルシステムのコンフィグレーション */
		private final DfsConfig config;
		/** 接続済みドメインコントローラー */
		private final DomainController dc;
		/** 現在のドメイン情報 */
		private Domain domain = null;
		/** ドメイン更新時に呼び出されるリスナー */
		private final ArrayList listeners = new ArrayList();

		/**
		 * コンストラクタ
		 * @param config_ ドメインファイルシステムのコンフィグレーション
		 * @param dc_ 接続済みドメインコントローラー
		 */
		public DomainSynchroThread(DfsConfig config_, DomainController dc_, Domain domain_) {
			config = config_;
			dc = dc_;
			domain = domain_;
			setDaemon(true);
		}
		/**
		 * スレッドメイン
		 */
		public void run() {
			try {
				while(true) {
					sleep(config.getDomainSynchroInterval());

					// ドメイン情報を取得する
					try {
						log.info("ドメイン情報を取得しています");
						Domain newDomain = domainController.getDomain(config.getDomainPreconnectId());
						if(domain.hashCode() != newDomain.hashCode()) {
							// ハッシュコードが変化したので、ドメイン情報が更新された
							DomainUpdateEventObject o = new DomainUpdateEventObject(this, domain, newDomain);
							executeCallbacks(o);

							// ドメイン情報を更新する
							domain = newDomain;
						} else {
							log.info("ドメイン情報は更新されていません");
						}
					} catch (RemoteException e1) {
						// 何らかの理由で、ドメインコントローラーからドメイン情報を取得できなかった
						// 一時的なことかもしれないので、無視する
						log.warn("ドメインコントローラーとの通信に失敗しました：", e1);
						log.warn("更新をスキップします");
						e1.printStackTrace();
					}
				}
			} catch(InterruptedException e) {
				log.info("*** ドメイン情報更新スレッドを停止しました");
			}
		}
		/**
		 * ドメイン情報更新スレッドを停止する
		 */
		public void shutdown() {
			interrupt();
		}
		/**
		 * イベントリスナーを起動する．
		 * @param o イベントオブジェクト
		 */
		protected void executeCallbacks(DomainUpdateEventObject o) {

			for(Iterator it = listeners.iterator(); it.hasNext();) {
				DomainUpdateListener lsnr = (DomainUpdateListener)it.next();
				lsnr.domainChanged(o);
			}
		}
		/**
		 * ドメイン更新通知リスナーを追加する．
		 * @param lsnr ドメイン更新通知リスナー
		 */
		public void addListener(DomainUpdateListener lsnr) {
			listeners.add(lsnr);
		}
	}
	/**
	 * クライアント接続モニタースレッド
	 * @author satoshi akabane@logical-paradox.org
	 * @version $Revision: 1.2 $
	 */
	class ClientConnectionMonitor extends Thread {
		/** ロガー */
		private final Log log = LogFactory.getLog(ClientConnectionMonitor.class);

		/** クライアント接続監視間隔(msec) */
		private final long monitoringInterval;

		/**
		 * コンストラクタ
		 * @param interval クライアント接続監視間隔(msec)
		 */
		public ClientConnectionMonitor(long interval) {
			monitoringInterval = interval;
			setDaemon(true);
		}
		public void run() {
			try {
				while(true) {
					sleep(monitoringInterval);

					ArrayList finalist = new ArrayList();

					synchronized(proxies) {
						for(Iterator it = proxies.iterator(); it.hasNext();) {
							DomainConnectionProxyImpl proxy = (DomainConnectionProxyImpl)it.next();
							if(proxy.isExpired() == true || proxy.isGarbageTarget() == true) {
								finalist.add(proxy);
							}
						}

						if(finalist.size() > 0) {
							log.info("全部で[" + finalist.size() + "]個のドメイン接続プロキシが破棄されます");

							// 回収対象を全て削除する
							for(Iterator it = finalist.iterator(); it.hasNext();) {
								DomainConnectionProxyImpl proxy = (DomainConnectionProxyImpl)it.next();
								proxies.remove(proxy);
								proxy.unshare();
							}
						}
					}
				}
			} catch(InterruptedException e) {
				log.info("クライアント接続モニターが停止しました");
			}
		}
	}
}
