/***************************************************************************
 *                                                                         *
 *                             ChordFuture.java                            *
 *                            -------------------                          *
 *   date                 : 16.08.2005                                     *
 *   copyright            : (C) 2004-2008 Distributed and                  *
 *                              Mobile Systems Group                       *
 *                              Lehrstuhl fuer Praktische Informatik       *
 *                              Universitaet Bamberg                       *
 *                              http://www.uni-bamberg.de/pi/              *
 *   email                : sven.kaffille@uni-bamberg.de                   *
 *   			    		karsten.loesing@uni-bamberg.de                 *
 *                                                                         *
 *                                                                         *
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   A copy of the license can be found in the license.txt file supplied   *
 *   with this software or at: http://www.gnu.org/copyleft/gpl.html        *
 *                                                                         *
 ***************************************************************************/
package de.uniba.wiai.lspi.chord.service.impl;

import static de.uniba.wiai.lspi.util.logging.Logger.LogLevel.DEBUG;
import static de.uniba.wiai.lspi.util.logging.Logger.LogLevel.INFO;

import java.io.Serializable;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import de.uniba.wiai.lspi.chord.com.CommunicationException;
import de.uniba.wiai.lspi.chord.com.Entry;
import de.uniba.wiai.lspi.chord.com.Node;
import de.uniba.wiai.lspi.chord.com.Proxy;
import de.uniba.wiai.lspi.chord.com.RefsAndEntries;
import de.uniba.wiai.lspi.chord.data.ID;
import de.uniba.wiai.lspi.chord.data.URL;
import de.uniba.wiai.lspi.chord.service.AsynChord;
import de.uniba.wiai.lspi.chord.service.Chord;
import de.uniba.wiai.lspi.chord.service.ChordCallback;
import de.uniba.wiai.lspi.chord.service.ChordFuture;
import de.uniba.wiai.lspi.chord.service.ChordRetrievalFuture;
import de.uniba.wiai.lspi.chord.service.Key;
import de.uniba.wiai.lspi.chord.service.Report;
import de.uniba.wiai.lspi.chord.service.ServiceException;
import de.uniba.wiai.lspi.util.logging.Logger;

/**
 * Implements all operations which can be invoked on the local node.
 * 
 * @author Karsten Loesing
 * @version 1.0.5
 */
public final class ChordImpl implements Chord, Report, AsynChord {

	/**
	 * Number of threads to allow concurrent invocations of asynchronous
	 * methods. e.g. {@link ChordImpl#insertAsync(Key, Serializable)}.
	 */
	private static final int ASYNC_CALL_THREADS = Integer.parseInt(System
			.getProperty(ChordImpl.class.getName() + ".AsyncThread.no"));

	/**
	 * Time in seconds until the stabilize task is started for the first time.
	 */
	private static final int STABILIZE_TASK_START = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.StabilizeTask.start"));

	/**
	 * Time in seconds between two invocations of the stabilize task.
	 */
	private static final int STABILIZE_TASK_INTERVAL = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.StabilizeTask.interval"));

	/**
	 * Time in seconds until the fix finger task is started for the first time.
	 */
	private static final int FIX_FINGER_TASK_START = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.FixFingerTask.start"));

	/**
	 * Time in seconds between two invocations of the fix finger task.
	 */
	private static final int FIX_FINGER_TASK_INTERVAL = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.FixFingerTask.interval"));

	/**
	 * Time in seconds until the check predecessor task is started for the first
	 * time.
	 */
	private static final int CHECK_PREDECESSOR_TASK_START = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.CheckPredecessorTask.start"));

	/**
	 * Time in seconds between two invocations of the check predecessor task.
	 */
	private static final int CHECK_PREDECESSOR_TASK_INTERVAL = Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.CheckPredecessorTask.interval"));

	/**
	 * Number of references in the successor list.
	 */
	private static final int NUMBER_OF_SUCCESSORS = (Integer
			.parseInt(System
					.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.successors")) < 1) ? 1
			: Integer
					.parseInt(System
							.getProperty("de.uniba.wiai.lspi.chord.service.impl.ChordImpl.successors"));

	/**
	 * Object logger.
	 */
	protected Logger logger;

	/**
	 * Reference on that part of the node implementation which is accessible by
	 * other nodes; if <code>null</code>, this node is not connected
	 */
	private NodeImpl localNode;

	/**
	 * Entries stored at this node, including replicas.
	 */
	private Entries entries;

	/**
	 * Executor service for local maintenance tasks.
	 */
	private ScheduledExecutorService maintenanceTasks;

	/**
	 * Executor service for asynch requests.
	 */
	private ExecutorService asyncExecutor;

	/**
	 * ThreadFactory used with Executor services.
	 * 
	 * @author sven
	 * 
	 */
	private static class ChordThreadFactory implements
			java.util.concurrent.ThreadFactory {

		private String executorName;

		ChordThreadFactory(String executorName) {
			this.executorName = executorName;
		}

		public Thread newThread(Runnable r) {
			Thread newThread = new Thread(r);
			newThread.setName(this.executorName + "-" + newThread.getName());
			return newThread;
		}

	}

	/**
	 * References to remote nodes.
	 */
	protected References references;

	/**
	 * Reference on hash function (singleton instance).
	 */
	private HashFunction hashFunction;

	/**
	 * This node's URL.
	 */
	private URL localURL;

	/**
	 * This node's ID.
	 */
	private ID localID;

	/* constructor */

	/**
	 * Creates a new instance of ChordImpl which initially is disconnected.
	 * Constructor is hidden. Only constructor.
	 */
	public ChordImpl() {
		this.logger = Logger.getLogger(ChordImpl.class.getName()
				+ ".unidentified");
		this.logger.debug("Logger initialized.");

		this.maintenanceTasks = new ScheduledThreadPoolExecutor(3,
				new ChordThreadFactory("MaintenanceTaskExecution"));
		this.asyncExecutor = Executors.newFixedThreadPool(
				ChordImpl.ASYNC_CALL_THREADS, new ChordThreadFactory(
						"AsynchronousExecution"));
		this.hashFunction = HashFunction.getHashFunction();
		logger.info("ChordImpl initialized!");
	}

	/**
	 * @return The Executor executing asynchronous request.
	 */
	final Executor getAsyncExecutor() {
		if (this.asyncExecutor == null) {
			throw new NullPointerException("ChordImpl.asyncExecutor is null!");
		}
		return this.asyncExecutor;
	}

	/* implementation of Chord interface */

	public final URL getURL() {
		return this.localURL;
	}

	public final void setURL(URL nodeURL) {

		if (nodeURL == null) {
			NullPointerException e = new NullPointerException(
					"Cannot set URL to null!");
			this.logger.error("Null pointer", e);
			throw e;
		}

		if (this.localNode != null) {
			IllegalStateException e = new IllegalStateException(
					"URL cannot be set after creating or joining a network!");
			this.logger.error("Illegal state.", e);
			throw e;
		}

		this.localURL = nodeURL;

		this.logger.info("URL was set to " + nodeURL);
	}

	public final ID getID() {
		return this.localID;
	}

	public final void setID(ID nodeID) {

		if (nodeID == null) {
			NullPointerException e = new NullPointerException(
					"Cannot set ID to null!");
			this.logger.error("Null pointer", e);
			throw e;
		}

		if (this.localNode != null) {
			IllegalStateException e = new IllegalStateException(
					"ID cannot be set after creating or joining a network!");
			this.logger.error("Illegal state.", e);
			throw e;
		}

		this.localID = nodeID;
		this.logger = Logger.getLogger(ChordImpl.class.getName() + "."
				+ this.localID);
	}

	public final void create() throws ServiceException {

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot create network; node is already connected!");
		}

		// has nodeURL been set?
		if (this.localURL == null) {
			throw new ServiceException("Node URL is not set yet!");
		}

		// if necessary, generate nodeID out of nodeURL
		if (this.getID() == null) {
			this.setID(this.hashFunction.createUniqueNodeID(this.localURL));
		}

		// establish connection
		this.createHelp();

	}

	public final void create(URL localURL1) throws ServiceException {

		// check if parameters are valid
		if (localURL1 == null) {
			throw new NullPointerException(
					"At least one parameter is null which is not permitted!");
		}

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot create network; node is already connected!");
		}

		// set nodeURL
		this.localURL = localURL1;

		// if necessary, generate nodeID out of nodeURL
		if (this.getID() == null) {
			this.setID(this.hashFunction.createUniqueNodeID(this.localURL));
		}

		// establish connection
		this.createHelp();

	}

	public final void create(URL localURL1, ID localID1)
			throws ServiceException {

		// check if parameters are valid
		if (localURL1 == null || localID1 == null) {
			throw new IllegalArgumentException(
					"At least one parameter is null which is not permitted!");
		}

		// is node already connected?
		if (this.localNode != null) {
			throw new ServiceException(
					"Cannot create network; node is already connected!");
		}

		// set nodeURL
		this.localURL = localURL1;