package com.dokukino.genkidama;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
//import java.util.Timer;
import java.util.concurrent.ExecutorService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.xmlrpc.XmlRpcException;
import org.positrium.ui.DHTSettingsDto;
import org.positrium.ui.TrayControl;
import org.seasar.framework.container.SingletonS2Container;

import ow.dht.ByteArray;
import ow.dht.DHT;
import ow.dht.DHTConfiguration;
import ow.dht.DHTFactory;
import ow.routing.RoutingException;

import dareka.common.Config;

public class DHTServer extends Thread {
	public static Logger logger = LoggerFactory.getLogger(DHTServer.class);
//	private Timer onlineUserUpdateTimer = new Timer();

	@SuppressWarnings("unchecked")
	public void connect() throws Exception {
		GenkidamaManager manager = GenkidamaManager.getInstance();
		int ringPort = manager.getRingPort();
		int dataPort = manager.getDataPort();
		String secret = Config.getString(
				"dhtSecret",
				null);
		assert secret != null;
		ByteArray hashedSecret = new ByteArray(secret.getBytes("UTF-8")).hashWithSHA1();

		String externalAddress = (String) XmlRpcInvoker.invoke(
				"globalIp.probe",
				null);
		manager.setExternalAddress(externalAddress);

		DHTConfiguration config = DHTFactory.getDefaultConfiguration();
		config.setRoutingStyle("Iterative");
		config.setRoutingAlgorithm("Kademlia");
		config.setMessagingTransport("TCP");
		config.setSelfPort(ringPort);
		config.setDoUPnPNATTraversal(true);
		config.setSelfAddress(externalAddress);
		config.setImplementationName("ChurnTolerantDHT");
		config.setDirectoryType("VolatileMap");
		config.setDoExpire(true);
		config.setDoReputOnReplicas(true);
		config.setDoReputOnRequester(true);

		DHT<String> dht = DHTFactory.getDHT(
				VersionInfo.APPLICATION_ID,
				VersionInfo.APPLICATION_MAJOR_VERSION,
				config,
				null);
		dht.setHashedSecretForPut(hashedSecret);

		manager.setDht(dht);

		List params = new ArrayList();
		params.add(ringPort);
		XmlRpcInvoker.invoke("tcpPort.check", params);
		{
			/* 依存性を解消しない代わりに、部分的にインジェクションを利用しつつSingletonでオブジェクトを得る */
			DHTSettingsDto app = SingletonS2Container.getComponent("dhtSettings");
			app.nodeIdentity = dht.getSelfIDAddressPair().getAddress().toString();
			app.transportType = config.getMessagingTransport();
			app.routingAlgorithm = config.getRoutingAlgorithm();
			app.routingStyle = config.getRoutingStyle();
			app.directoryType = config.getDirectoryType();
			app.workingDirectory = config.getWorkingDirectory();

			Object[] msg = {
					app.nodeIdentity,
					app.transportType,
					app.routingAlgorithm,
					app.routingStyle,
					app.directoryType,
					app.workingDirectory };

			logger.info(
					app.log_msg_format,
					msg);
		}

		// join 出来るまでループし続ける
		joinLoop: while (true) {
			params = new ArrayList();
			params.add(ringPort);
			params.add(VersionInfo.APPLICATION_MINOR_VERSION);
			Object[] onlineUsers = (Object[]) XmlRpcInvoker.invoke(
					"onlineUsers.fetch",
					params);
			if (onlineUsers.length == 0) {
				logger.info("DHT started as standalone");
				break;
			}
			for (int i = 0; i < onlineUsers.length; i++) {
				HashMap user = (HashMap) onlineUsers[i];
				String address = (String) user.get("address");
				Integer port = (Integer) user.get("port");
				if (address.equals(externalAddress) && port == ringPort)
					continue;

				String url = address + ":" + port;
				logger.info(
						"DHT joining to {}",
						url);

				try {
					dht.joinOverlay(url);
					logger.info(
							"DHT joined via {}",
							url);
					break joinLoop;
				} catch (RoutingException e) {
					{
						Object[] msg = {
								url,
								i + 1,
								onlineUsers.length };
						logger.warn(
								"can not connect node {} : {} of {}",
								msg);
					}

				} catch (Exception e) {
					if (logger.isErrorEnabled())
						logger.error(
								"Error Occurs: ",
								e);
				}
			}
		}

		if (Config.getBoolean(
				"dhtShell",
				false)) {
			DHTShellServer.start(dht);
		}

		params = new ArrayList();
		params.add(ringPort);
		params.add(VersionInfo.APPLICATION_MINOR_VERSION);
		XmlRpcInvoker.invoke(
				"onlineUsers.register",
				params);
/*
		onlineUserUpdateTimer.scheduleAtFixedRate(
				new OnlineUserUpdateTask(),
				0,
				OnlineUserUpdateTask.TASK_PERIOD);
*/
		params = new ArrayList();
		params.add(dataPort);
		XmlRpcInvoker.invoke(
				"tcpPort.check",
				params);
		manager.setConnected(true);
		ExecutorService putExecutor = manager.getPutExecutor();
		synchronized (putExecutor) {
			if (logger.isDebugEnabled())
				logger.debug("notifying connected");
			putExecutor.notifyAll();
		}
	}

	@SuppressWarnings("unchecked")
	public void disconnect() throws Exception {
		GenkidamaManager manager = GenkidamaManager.getInstance();
		ExecutorService putExecutor = manager.getPutExecutor();
		putExecutor.shutdown();
//		onlineUserUpdateTimer.cancel();
		List params = new ArrayList();
		int ringPort = manager.getRingPort();
		params.add(ringPort);
		params.add(VersionInfo.APPLICATION_MINOR_VERSION);
		try {
			XmlRpcInvoker.invoke(
					"onlineUsers.unregister",
					params);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			logger.error(
					"Error Occurs:",
					e);
		}
		DHT<String> dht = manager.getDht();
		dht.stop();
		logger.info("DHT network disconnected");
		manager.setConnected(false);
	}

	public void run() {
		TrayControl tray = SingletonS2Container.getComponent(TrayControl.class);

		try {
			connect();
			/*
			 * synchronized(this) { wait(); } disconnect();
			 */
		} catch (XmlRpcException e) {
			switch (e.code) {
			case XmlRpcError.PORT:
				// TODO Auto-generated catch block
				if (logger.isErrorEnabled())
					logger.error(
							"Error Occurs:",
							e);
				tray.fatalStop(
						"Error",
						"Port open check failed; Please check network settings.");
				break;
			default:
				// TODO Auto-generated catch block
				if (logger.isErrorEnabled())
					logger.error(
							"Error Occurs:",
							e);
				tray.fatalStop(
						"Error",
						e.toString());
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			if (logger.isErrorEnabled())
				logger.error(
						"Error Occurs:",
						e);
			tray.fatalStop(
					"Error",
					e.toString());
		}
	}
}
