package com.dokukino.genkidama;

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

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.positrium.chikarawo.ErrorDialog;

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 Log logger = LogFactory.getLog(DHTServer.class);
	
	@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.setMessagingTransport("TCP");
		config.setRoutingAlgorithm("Kademlia");
		config.setRoutingStyle("Iterative");
		config.setDoUPnPNATTraversal(false);
		config.setSelfPort(ringPort);
		config.setSelfAddress(externalAddress);
		config.setDoExpire(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);

		StringBuilder sb = new StringBuilder();
		sb.append("DHT configuration:\n");
		sb.append("  hostname:port:     ").append(
				dht.getSelfIDAddressPair().getAddress()).append('\n');
		sb.append("  transport type:    ").append(
				config.getMessagingTransport()).append('\n');
		sb.append("  routing algorithm: ").append(
				config.getRoutingAlgorithm()).append('\n');
		sb.append("  routing style:     ").append(config.getRoutingStyle())
				.append('\n');
		sb.append("  directory type:    ")
				.append(config.getDirectoryType()).append('\n');
		sb.append("  working directory: ").append(
				config.getWorkingDirectory()).append('\n');
		logger.info(sb);

		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) {
				if (logger.isInfoEnabled())
					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;
				if (logger.isInfoEnabled())
					logger.info("DHT joining to " + url);
				try {
					dht.joinOverlay(url);
					if (logger.isInfoEnabled())
						logger.info("DHT joined via " + url);
					break joinLoop;
				} catch (RoutingException e) {
					logger.error("Error Occurs: ", e);
					if (logger.isInfoEnabled()) {
						StringBuilder stringbuild = new StringBuilder(
								"can not connect node (perhaps, invalid node data) : ");
						stringbuild.append(url);
						stringbuild.append(" : ");
						stringbuild.append(i + 1); // for logger.info
						stringbuild.append(" of ");
						stringbuild.append(onlineUsers.length);
						if ((i + 1) == onlineUsers.length) {
							stringbuild.append(" retry forever..");
						}
						stringbuild.trimToSize();
						logger.info(stringbuild.toString());
					}
				} 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);

		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();
		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() {
		try {
			connect();
			/*
			 * synchronized(this) { wait(); } disconnect();
			 */
		} catch (Exception e) {
			// TODO Auto-generated catch block
			if (logger.isErrorEnabled())
				logger.error("Error Occurs:", e);
			new ErrorDialog("Error", e.toString());
			System.exit(1);
		}
	}
}
