/*
 * Copyright (c) 2009, syuu
 * Licensed under the NicoCache License.
 */

package com.dokukino.genkidama;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.net.HttpURLConnection;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import ow.dht.ByteArray;
import ow.dht.DHT;
import ow.dht.DHTConfiguration;
import ow.dht.DHTFactory;
import ow.dht.ValueInfo;
import ow.id.ID;
import ow.routing.RoutingException;
import ow.tool.dhtshell.commands.ClearCommand;
import ow.tool.dhtshell.commands.GetCommand;
import ow.tool.dhtshell.commands.HaltCommand;
import ow.tool.dhtshell.commands.HelpCommand;
import ow.tool.dhtshell.commands.InitCommand;
import ow.tool.dhtshell.commands.LocaldataCommand;
import ow.tool.dhtshell.commands.PutCommand;
import ow.tool.dhtshell.commands.QuitCommand;
import ow.tool.dhtshell.commands.RemoveCommand;
import ow.tool.dhtshell.commands.ResumeCommand;
import ow.tool.dhtshell.commands.SetSecretCommand;
import ow.tool.dhtshell.commands.SetTTLCommand;
import ow.tool.dhtshell.commands.StatusCommand;
import ow.tool.dhtshell.commands.SuspendCommand;
import ow.tool.util.shellframework.Command;
import ow.tool.util.shellframework.MessagePrinter;
import ow.tool.util.shellframework.Shell;
import ow.tool.util.shellframework.ShellServer;

import dareka.common.Config;
import dareka.processor.HttpResponseHeader;
import dareka.processor.URLResource;
import dareka.processor.impl.NicoApiUtil;

public class DHTManager {
	public static Log logger = LogFactory.getLog(DHTManager.class);
	private boolean connected;
	DHT<String> dht;
	private static DHTManager instance = new DHTManager();
	String externalAddress;
	ByteArray hashedSecret;
	private int ringPort, dataPort;

	private DataServer dataServer;
	private Map<ID, String> distributedMap = new ConcurrentHashMap<ID, String>();

	private BlockingQueue<String> putQueue = new LinkedBlockingQueue<String>();

	private DHTMainThread mainThread = new DHTMainThread();
	
	private final static int SHELL_PORT = -1;
	private final static Class/* Command<<DHT<String>>> */[] COMMANDS = {
			StatusCommand.class, InitCommand.class, GetCommand.class,
			PutCommand.class, RemoveCommand.class, SetTTLCommand.class,
			SetSecretCommand.class, LocaldataCommand.class,
			// SourceCommand.class,
			HelpCommand.class, QuitCommand.class, HaltCommand.class,
			ClearCommand.class, SuspendCommand.class, ResumeCommand.class };

	private final static List<Command<DHT<String>>> commandList;
	private final static Map<String, Command<DHT<String>>> commandTable;

	static {
		commandList = ShellServer.createCommandList(COMMANDS);
		commandTable = ShellServer.createCommandTable(commandList);
	}

	private DHTManager() {
	}

	public static DHTManager getInstance() {
		return instance;
	}

	public int getRingPort() {
		return ringPort;
	}

	public void connect() throws Exception {
		mainThread.start();
	}

	public void disconnect() throws RoutingException {
		/*
		synchronized(mainThread) {
			mainThread.notify();
		}
		*/
		try {
			if(connected)
				mainThread.disconnect();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			if (logger.isErrorEnabled())
				logger.error("Error Occurs:", e);
		}
	}

	public boolean isConnected() {
		return connected;
	}

	public String getNicoCache(String id) throws RoutingException {
		if (!checkDeleted(id)) {
			ID key = ID.getSHA1BasedID(("nc:" + id).getBytes());
			Set<ValueInfo<String>> uriList = dht.get(key);
			for (ValueInfo<String> v : uriList) {
				String uri = v.getValue();
				try {
					java.net.URL url = new java.net.URL(uri);
					if (url.getHost().equals(externalAddress))
						continue;
					HttpURLConnection conn = (HttpURLConnection) url
							.openConnection();
					conn.setRequestMethod("HEAD");
					conn.setConnectTimeout(Config
							.getInteger("dhtTimeout", 1000));
					conn.connect();
					conn.disconnect();
					if (logger.isInfoEnabled())
						logger.info("DHT get: nc:" + id + " success");
					return uri;
				} catch (Exception e) {
					if (logger.isDebugEnabled())
						logger.debug(e);
				}
			}
		}
		logger.info("DHT get: nc:" + id + " failed");
		return null;
	}

	public void putNicoCache(String id) throws Exception {
		putQueue.put(id);
	}

	public static boolean checkDeleted(String id) {
		id = id.replaceAll("low$", "");
		String title = null;
		try {
			String url = NicoApiUtil.getThumbURL(id);
			URLResource r = new URLResource(url);
			// In the general case, proxy must let browser know a redirection,
			// but in this case, behave just as a client.
			r.setFollowRedirects(true);

			ByteArrayOutputStream bout = new ByteArrayOutputStream();
			r.transferTo(null, bout, null, null);

			ByteArrayInputStream bin = new ByteArrayInputStream(bout
					.toByteArray());
			new HttpResponseHeader(bin); // skip header

			title = NicoApiUtil.getThumbTitle(bin);

			bout.close();
			bin.close();
		} catch (Exception e) {
			logger.error(e);
			e.printStackTrace();
		}
		if (title == null) {
			logger.info("DHT cacheDeleted: " + id + " -> true");
			return true;
		} else {
			logger.info("DHT cacheDeleted: " + id + " -> false");
			return false;
		}
	}

	private static class ShowPromptPrinter implements MessagePrinter {
		public void execute(PrintStream out, String hint) {
			out.print("Ready." + Shell.CRLF);
			out.flush();
		}
	}

	private static class NoCommandPrinter implements MessagePrinter {
		public void execute(PrintStream out, String hint) {
			out.print("No such command");

			if (hint != null)
				out.print(": " + hint);
			else
				out.print(".");
			out.print(Shell.CRLF);

			out.flush();
		}
	}

	private class DHTMainThread extends Thread {
		public void connect() throws Exception {
			ringPort = Config.getInteger("dhtRingPort", 9899);
			dataPort = Config.getInteger("dhtDataPort", 9900);
			assert ringPort != dataPort;
			String secret = Config.getString("dhtSecret", null);
			assert secret != null;
			hashedSecret = new ByteArray(secret.getBytes("UTF-8"))
					.hashWithSHA1();
			externalAddress = (String) XmlRpcInvoker.invoke(
					"globalIp.probe", null);

			DHTConfiguration config = DHTFactory.getDefaultConfiguration();
			config.setMessagingTransport("TCP");
			config.setRoutingAlgorithm("Chord");
			config.setRoutingStyle("Iterative");
			config.setDoUPnPNATTraversal(false);
			config.setSelfPort(ringPort);
			config.setSelfAddress(externalAddress);

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

			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) {
						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);
					}
				}
			}

			DHTPutThread putThread = new DHTPutThread(putQueue);
			putThread.start();

			if (Config.getBoolean("dhtShell", false)) {
				// start a ShellServer
				ShellServer<DHT<String>> shellServ = new ShellServer<DHT<String>>(
						commandTable, commandList, new ShowPromptPrinter(),
						new NoCommandPrinter(), null, dht, SHELL_PORT, null);

				Shell<DHT<String>> stdioShell = new Shell<DHT<String>>(
						System.in, System.out, shellServ, dht, true);
				Thread shellThread = new Thread(stdioShell);
				shellThread.start();
			}

			params = new ArrayList();
			params.add(ringPort);
			params.add(VersionInfo.APPLICATION_MINOR_VERSION);
			XmlRpcInvoker.invoke("onlineUsers.register", params);

			dataServer = new DataServer();
			dataServer.bind(dataPort);

			params = new ArrayList();
			params.add(dataPort);
			XmlRpcInvoker.invoke("tcpPort.check", params);
			connected = true;
			
		}
		
		public void disconnect() throws Exception {
			for (Entry<ID, String> entry : distributedMap.entrySet()) {
				String key = entry.getValue().replaceFirst(
						"^http://.+:[0-9]+/nc/", "nc:");
				if (logger.isInfoEnabled())
					logger.info("DHT remove: " + key);
				List params = new ArrayList();
				params.add(ringPort);
				params.add(VersionInfo.APPLICATION_MINOR_VERSION);
				params.add(key);
				try {
					XmlRpcInvoker.invoke("onlineData.unregister", params);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					if (logger.isErrorEnabled())
						logger.error("Error Occurs:", e);
				}
				dht.remove(entry.getKey(), hashedSecret);
			}
			List params = new ArrayList();
			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.stop();
			logger.info("DHT network disconnected");
			connected = 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);
				System.exit(1);
			}
		}
	}

	private class DHTPutThread extends Thread {
		private BlockingQueue<String> queue;

		private DHTPutThread(BlockingQueue<String> queue) {
			this.queue = queue;
		}

		@SuppressWarnings("unchecked")
		public void run() {
			while (true) {
				try {
					String id = queue.take();
					if (logger.isDebugEnabled())
						logger.debug("PutThread takes " + id);
					if (!checkDeleted(id)) {
						ID key = ID.getSHA1BasedID(("nc:" + id).getBytes());
						logger.info("DHT put: nc:" + id);
						String uri = "http://" + externalAddress + ":"
								+ dataPort + "/nc/" + id;
						distributedMap.put(key, uri);
						try {
							dht.put(key, uri);
							List params = new ArrayList();
							params.add(ringPort);
							params.add(VersionInfo.APPLICATION_MINOR_VERSION);
							params.add("nc:" + id);
							XmlRpcInvoker.invoke("onlineData.register", params);
						} catch (Exception e) {
							// TODO Auto-generated catch block
							if (logger.isErrorEnabled())
								logger.error("Error Occurs:", e);
						}
					}
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					if (logger.isErrorEnabled())
						logger.error("Error Occurs:", e);
				}

			}
		}
	}
}
