package de.fu.alp5.foreign.distributed;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.StreamTokenizer;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.rmi.AlreadyBoundException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;

/**
 * Analyzes a text for foreign words.
 * 
 * @author Konrad Reiche
 * 
 */
public class TextAnalyzerImpl implements TextAnalyzer {

	List<String> text;
	List<String> dictionary;
	List<String> foreignDictionary;
	CopyOnWriteArraySet<String> globalForeignWordSet;

	/**
	 * Converts the given text files to lists of Strings. Binds and export the
	 * current instance to the RMI registry in order to be accessible. If the
	 * TextAnalyzer is already bound ignore the exceptions as the current
	 * instance is still exported.
	 */
	public TextAnalyzerImpl(String textPath, String dictionaryPath,
			String foreignDictionaryPath) throws IOException {

		text = convertTextFileToList(textPath);
		dictionary = convertTextFileToList(dictionaryPath);
		foreignDictionary = convertTextFileToList(foreignDictionaryPath);
		globalForeignWordSet = new CopyOnWriteArraySet<String>();

		TextAnalyzer stub = (TextAnalyzer) UnicastRemoteObject
				.exportObject((TextAnalyzer) this);

		try {
			LocateRegistry.getRegistry().bind("TextAnalyzer", stub);
		} catch (AlreadyBoundException e) {
			// ignore
		}
	}

	/**
	 * Starts the analyzing. A CountDownLatch is used to synchronize the
	 * termination of the method call. The parameter numberOfThreads determines
	 * the number of TextAnaylzerWorker to be spawned.
	 * 
	 * A list of the worker host is passed in order to look up every worker
	 * object. When the worker object is retrieved it is initialized.
	 * 
	 * For every Worker a new Thread is spawned and an anonymous Runnable is
	 * instantiated because the run method cannot be invoked remotely.
	 * 
	 * When the analyzing is finished the instance of this object is unbound and
	 * unexported.
	 */
	public void analyze(int numberOfThreads, List<String> hostList)
			throws RemoteException, UnknownHostException, InterruptedException {

		final CountDownLatch countDownLatch = new CountDownLatch(
				numberOfThreads);

		for (int i = 0; i < numberOfThreads; ++i) {
			Registry registry = LocateRegistry.getRegistry(hostList.get(i));
			final TextAnalyzerWorker worker;

			try {
				worker = (TextAnalyzerWorker) registry
						.lookup("TextAnalyzerWorker");
			} catch (NotBoundException e1) {
				System.err.println("There is no TextAnalyzerWorker on "
						+ hostList.get(i));
				continue;
			}

			worker.initialize(numberOfThreads, i, InetAddress.getLocalHost()
					.getHostAddress());

			new Thread(new Runnable() {

				@Override
				public void run() {

					try {
						worker.analyze();
						countDownLatch.countDown();
					} catch (RemoteException e) {
						System.err.println(e);
					}

				}
			}).start();
		}

		countDownLatch.await();

		try {
			LocateRegistry.getRegistry().unbind("TextAnalyzer");
		} catch (NotBoundException e) {
			System.err.println("TextAnalyzer was not bound");
		}

		UnicastRemoteObject.unexportObject((TextAnalyzer) this, true);
	}

	/**
	 * Converts an arbitrary text file to a list of Strings by using
	 * StreamTokenizer.
	 */
	public static List<String> convertTextFileToList(String path)
			throws IOException {

		List<String> result = new ArrayList<String>();
		FileInputStream textInputStream = new FileInputStream(path);
		StreamTokenizer streamTokenizer = new StreamTokenizer(
				new InputStreamReader(textInputStream));
		streamTokenizer.lowerCaseMode(true);

		int tokenByte = streamTokenizer.nextToken();
		while (tokenByte != StreamTokenizer.TT_EOF) {
			if (tokenByte == StreamTokenizer.TT_WORD) {
				result.add(streamTokenizer.sval);
			}
			tokenByte = streamTokenizer.nextToken();
		}

		return result;
	}

	/**
	 * Returns a String with all foreign words found in the given text.
	 */
	@Override
	public String toString() {
		String result = new String();

		for (String foreignWord : globalForeignWordSet) {
			result = result.concat(foreignWord + "\n");
		}

		return result;
	}

	@Override
	public void addForeignWord(String foreignWord) {
		globalForeignWordSet.add(foreignWord);
	}

	@Override
	public CopyOnWriteArraySet<String> getGlobalForeignWordSet() {
		return globalForeignWordSet;
	}
}