package taskspaces.system;

import java.io.*;
import java.net.*;
import java.util.*;
import java.security.cert.*;
import javax.net.ssl.*;

/**
 * Represents a server thread and a set of data structures and methods
 * for sending, storing, and retrieving executable objects between
 * applications and Nodes. 
 * @author Rob Markel, Matthew Keoshkerian
 * @version 2.0 
 */
public class Space extends Communicator implements Runnable
{
	/** Holds the tasks waiting to be sent out to nodes. */
	public static List tasks=Collections.synchronizedList(new ArrayList());
	
	/** Holds the list of all the nodes which are connected, not just idle. */
	public static List nodesPermanent=Collections.synchronizedList(new ArrayList());
	
	/** Holds messages which would be transferred to nodes. */
	public static Map messages=Collections.synchronizedMap(new HashMap());
	
	/** Holds the keys to the messages. */
	public static Map keys=Collections.synchronizedMap(new HashMap());
	
	/** Holds the list of all the Idle nodes waiting for jobs. */
	public static List nodes=Collections.synchronizedList(new ArrayList());
    
	/** The ServerSocket which will accept all the incomming connections. */
	private ServerSocket s;
	
	/** The URLClassLoader of this class. */
	protected URLClassLoader classLoaderPath;
   
	// The list of URLS that need to be loaded.
	//URL [] urls;

	/**
	 * Starts a new Space in a server thread running on an open port. The
	 * thread remains running waiting for connections to store or send 
	 * data and executables.
	 * @return void
	 */
	public void run()
	{
		try
		{
			(new File("./logs/")).mkdir();
			logWriteName="./logs/system.system";
			logPrefix = "Space" ;
			logWrite = new PrintWriter(new FileWriter(logWriteName, true));
			classLoaderPath = (URLClassLoader)this.getClass().getClassLoader();
			
			// Retrieve the address of the properties file from the System. 
			String propertiesAddress=System.getProperty("propsAddress");
			
			// Load the properties file (NB: using the load algorithm from the Communicator clas)
			Properties p=load(propertiesAddress);
			
			// Take the port this Space is to listen on and have the server socket listen on it. If it is already in use, incriment the port number and try again.
			int port=Integer.parseInt(p.getProperty("space_port"));
			//print("space_port "+port);
			
			while(s==null)
			{
				// Try creating a ServerSocket on the initial port, and if unable to, try the next port up.
				try {s=new ServerSocket(port);} 
				catch(BindException e) 
				{ 
					port++;  
					System.setProperty("port",String.valueOf(port));
				}
			}
			print("Space: Running on port "+port);
			
			
			// Retrieve the spaces property from the properties file.
			String propertySpaces=p.getProperty("spaces");
			
			// Parse the property and assign it to the spaces array.
			String[] spaces=parseProperty(propertySpaces);
			
			// Get the host name and all IP addresses assigned to this hostname and set the IP/port variables this machine is assigned/listening to.
			String host=InetAddress.getLocalHost().getHostName();
			InetAddress[] i=InetAddress.getAllByName(host);
			String thisIP=i[0].getHostAddress();
			String thisPort=String.valueOf(port);
			boolean match=false;
			
			// Parse the spaces property and loop through to see if this IP/port match the property file's IP/port specification.
			for(int j=0; j<spaces.length; j++)
			{
				StringTokenizer tokenizer=new StringTokenizer(spaces[j],":");
				String propertyIPAddress=tokenizer.nextToken();
				String propertyPort=tokenizer.nextToken();
				
				print("Space: properties file  "+propertyPort+" "+propertyIPAddress+" || actual "+thisPort+" "+thisIP);

				/*if(thisPort.equals(propertyPort) && thisIP.equals(propertyIPAddress))
				{
					match=true; 
				}*/
				/*if(thisPort.equals(propertyPort)) 
				{
					thisIP=propertyIPAddress; 
				}*/
				
			}
			/*if(!match)
			{
				// If the IP and port do not match, exit out of the system (disabled for now).
				print("IP address and port do not match any specified in properties file!");
				//   System.exit(0);
			}*/
			// End of check.

			// Create a new Thread group for all incoming connections.
			ThreadGroup tg=new ThreadGroup("Connections");
			//tg.setMaxPriority(this.getPriority()-1);
         
			// Wait for a connection attempt. Once a user tries to connect, place him to the SpaceConnection and wait again for another user to connect.
			while(true)
			{
				Socket socket=s.accept();
				//Class c = classLoaderPath.loadClass("SpaceConnection");
				//Class [] constructorParams = new Class[] {tg.getClass(), s.getClass(), this.getClass()};
				//Object [] initargs = new Object[] {tg, s, this};
				//((Runnable)(c.getDeclaredConstructor(constructorParams)).newInstance(initargs)).run();
				SpaceConnection c=new SpaceConnection(tg,socket,this);
			}
		}
		catch(Exception e) {e.printStackTrace();}
	}

	/**
	 * Sends an Object to a registered Node.
	 * @param o an instantiated subclass of Task.
	 * @return <code>true</code> if send is successful.
	 */
	public boolean sendOut(Object o)
	{
		boolean sent=false;
		while(nodes.size()>0)
		{
			String address=(String)nodes.remove(0);
			if(address!=null)
			{
				sent=sendToNode(address,o);
				if(sent) return sent;
			}
		}
		return sent;
	}
}

/**
 * Represents a client connection to the Space, started in
 * a separate Thread so the parent Space can continue to handle requests.
 * @author Rob Markel, Matthew Keoshkerian
 * @version 2.0
 */
class SpaceConnection extends Thread implements javax.net.ssl.X509TrustManager
{
	private static int counter=0;
	protected ObjectInputStream in;
	protected ObjectOutputStream out;
	protected Socket client;
	protected String ip;
	protected int port;
	protected String clientAddress;
	protected Space space;
	Communicator communicator=new Communicator();

    
	public SpaceConnection() {}
	/**
	 * Constructor for the SpaceConnection class
	 * @param group The ThreadGroup this connection will belong to
	 * @param client The Socket connection for the client
	 * @param space The refrence to the Space which this client originated from.
	 */
	public SpaceConnection(ThreadGroup group,Socket client,Space space)
	{
		// Create an instance of the superclass, Thead, using the ThreadGroup passed through.
		super(group,"Connection "+counter++);
      
		// Try to set all the class variables. If someone goes wrong with the connection, it will be terminated.
		try
		{
			// Set the client, IP of the client, the port the client is connected on, set it's address, and get it's input stream.
			this.client=client;
			ip=client.getInetAddress().getHostAddress();
			port=client.getPort();
			clientAddress=ip+":"+port; 
			in=new ObjectInputStream(client.getInputStream());
	 
			// Set the space.
			this.space=space;
		}
		catch(IOException e)
		{
			// An error occured, close the stream.
			try {client.close();}
			catch(IOException ioe) {}
			return;
		}
      
		// Start the process. Invokes the run method.
		start();
	}
	
	private void print (String line)
	{
		space.print(line);
	}

	/**
	 * Handles the client request.
	 * @return void
	 */
	public void run()
	{
		Object o;
      
      
		try
		{  
			//SSLContext sc = SSLContext.getInstance("SSLv3");
			//TrustManager[] tma = { new SpaceConnection() };
			//sc.init(null, tma, null);
			
			// Wait for the client to send a request, and deal with the request accordingly.
			switch(in.readInt())
			{
            	// Run agent
            	case Space.AGENT:
            	{
            		Agent a=(Agent)in.readObject();
            		a.setSpace(space);
            		if (a instanceof LogAgent)
            			((LogAgent)a).setVars(client, space); 
            		a.run();
            		break;
            	}
	    
            	// Load one URL
            	case Space.URLLOADER:
            	{
            		try {
            			//ClassLoader cload = space.getClass().getClassLoader();
            			//System.out.println(cload instanceof URLClassLoader);
            			//System.out.println(cload instanceof URLClassLoaderExt);
            			//URLClassLoader ucload = (URLClassLoader)cload;
            			//URLClassLoaderExt over = (URLClassLoaderExt)ucload;
            			//overl.setLoad(ucload);
            			//over.addURLPath((URL)in.readObject());
		    
            			//URL [] urls = new URL[] {(URL)in.readObject()};
            			//space.classLoaderPath = new URLClassLoader(urls, space.classLoaderPath);
		    
            			URL add = (URL)in.readObject();
            			space.addURL(add, space);
            			Object [] nodes = space.nodesPermanent.toArray();
		    
            			for (int i = 0; i < nodes.length; i++)
            			{
            				space.sendToNode((String)nodes[i], add);
            			}

		    
            			//ucload.addURL((URL)in.readObject());
            			//space.urls = new URL [2]; 
            			//space.urls[0] = (URL)in.readObject();
            			//space.urls[1] = new URL("http://alphacelestia.math.uwaterloo.ca/");
            			//System.out.println("URL added");
            			//URLClassLoader u = new URLClassLoader(url);
            		}
            		catch(Exception e) { e.printStackTrace(); }
            		break;
            	}
	    
            	// Load many URLs
            	case Space.URLLOADERMANY:
            	{
            		URL [] urls = (URL[])in.readObject();
            		for (int i = 0; i < urls.length; i++)
            		{
            			space.addURL(urls[i], space);
            		}
            		//space.classLoaderPath = new URLClassLoader(urls, space.classLoaderPath);
            		//URLClassLoader u = new URLClassLoader(urls);
            		break;
            	}
 
            	// Accepts many objects over same socket
            	case Space.IN_MANY:
            	{
            		int num=in.readInt();
            		for(int i=0; i<num; i++)
            		{
            			o=in.readObject();
            			if(Space.nodes.size() > 0) 
				    space.sendOut(o);
            			else 
				    Space.tasks.add(o);
            			print("Space: Tasks size: "+Space.tasks.size());
            			print("Space: Nodes size: "+Space.nodes.size());
            		} 
            		break;
            	}

            	/*
            	 case Space.OUT_MANY:
            	 System.out.println("Received OUTMANY Request");
            	 int numToSend=in.readInt();
            	 System.out.println("NumToSend: "+numToSend);
            	 out=new ObjectOutputStream(client.getOutputStream());
            	 int count=0;
            	 for(int j=0; j<numToSend;)
            	 {
            	 System.out.println("Loop Space Size: "+Space.tasks.size());
            	 System.out.println("Count: "+count);
            	 if(Space.tasks.size()>0) 
            	 {
                  	o=Space.tasks.remove(0);
                  	out.writeObject(o);
                  	j++; 
                  	System.out.println("Space size: "+Space.tasks.size());
                 } 
                 }
                  	break;
            */ 

            	// Accepts one object per socket
            	case Space.IN:
            	{
            		o=in.readObject(); 
            		if(Space.nodes.size()>0) 
            			space.sendOut(o);
            		else
            			Space.tasks.add(o); 
            		print("Space: Tasks size: "+Space.tasks.size());
            		print("Space: Nodes size: "+Space.nodes.size());
            		break;
            	}

            	// Sends one object per socket
            	case Space.OUT:
            	{
            		out=new ObjectOutputStream(client.getOutputStream());
            		if(Space.tasks.size()>0) 
            			out.writeObject(Space.tasks.remove(0));
            		out.reset();
            		print("Space: Space size: "+Space.tasks.size());
            		break;
            	}


            	/*--== Message handling section ==--*/

            	// Store a message.
            	case Space.STORE_MESSAGE:
            	{
            		String key=(String)in.readObject();
            		int _port=in.readInt();
            		String toAddress=ip+":"+_port; 
            		if(Space.keys.containsKey(key))
            		{
            			String sendToAddress=(String)Space.keys.remove(key);
            			print("Space: Sending message to "+sendToAddress);
            			communicator.sendDirectMessage(sendToAddress,key,toAddress); 
            		}
            		else {Space.messages.put(key,toAddress);}
            		break;
            	}

            	// Send a mesage
            	case Space.SEND_MESSAGE:
            	{
            		String aKey=(String)in.readObject();
            		int _prt=in.readInt();
            		String sendToAddress=ip+":"+_prt;
            		if(Space.messages.containsKey(aKey))
            		{
            			String requestedAddress=(String)Space.messages.remove(aKey);
            			print("Space: Sending message to "+sendToAddress);
            			communicator.sendDirectMessage(sendToAddress,aKey,requestedAddress);
            		}
            		else {Space.keys.put(aKey,sendToAddress);}
            		break;
            	}

            	// Reset the Space.
            	case Space.RESET:
            	{
            		print("Space: Resetting.");
            		Space.tasks.clear();
            		Space.messages.clear();
            		Space.keys.clear();
            		Space.nodes.clear();
            		Space.nodesPermanent.clear();
            		break;
            	}

            	/*--== WORKERS ==-- */   

            	case Space.REGISTER_NODE:
            	{
            		int workerPort=in.readInt();
            		print("Space: Received worker registration from: "+ip+":"+workerPort);
            		Space.nodes.add(ip+":"+workerPort);
            		boolean match=false; 
            		for(int i=0; i<Space.nodesPermanent.size(); i++)
            		{
            			String address=(String)Space.nodesPermanent.get(i);
            			if(address.equals((ip+":"+workerPort))) 
            			{
            				match=true;
            			}
            		}
            		if(match==false) 
            		{
            			Space.nodesPermanent.add(ip+":"+workerPort);
            		}
            		if(Space.tasks.size()>0) {boolean b=sendOut(Space.tasks.remove(0));}
            		print("Space: Tasks size: "+Space.tasks.size());
            		print("Space: Nodes size: "+Space.nodes.size());
            		print("Space: Permanent Nodes size: "+Space.nodesPermanent.size());
            		break;
            	}


            	case Space.DELETE_NODE:
            	{
            		print("Space: Received worker delete from: "+clientAddress);
            		int deletePort=in.readInt();
            		Space.nodes.remove(ip+":"+deletePort);
            		Space.nodesPermanent.remove(ip+":"+deletePort);
            		break;
            	}

            	/*--== QUERY SECTION ==--*/

            	// Return the number of nodes.
            	case Space.NODE_QUERY:
            	{
            		out=new ObjectOutputStream(client.getOutputStream());
            		out.writeInt(Space.nodes.size());
            		break;
            	}

            	// Return an array of Idle nodes
            	case Space.GET_IDLE_NODES:
            	{
            		print("Space: Received request for worker addresses.");
            		out=new ObjectOutputStream(client.getOutputStream());
            		out.writeObject(Space.nodes.toArray());
            		break;
            	}

            	// Return the array of all the Nodes.
            	case Space.GET_ALL_NODES:
            	{
            		print("Space: Received request for worker addresses.");
            		out=new ObjectOutputStream(client.getOutputStream());
            		out.writeObject(Space.nodesPermanent.toArray());
            		out.flush();
            		break;
            	}
			}
			// Close the input stream and output stream (if any).
			in.close();
			print("Space: Closing...");
			if(out!=null) {out.close();}
			// Close the client's connection.
			client.close();
		} catch(Exception e) {e.printStackTrace(); return;}
	}

	/**
	 * Sends an executable Task to a node.
	 * @param o a subclass of Task.
	 * @return <code>true</code> if successful.
	 */
	public boolean sendOut(Object o)
	{
		boolean sent=false;
		while(Space.nodes.size()>0)
		{
			String address=(String)Space.nodes.remove(0);
			if(address!=null)
			{
				sent=communicator.sendToNode(address,o);
				if(sent) return sent;
			}
		}
		return sent;

	}
    
	/** Required methods for the trust manager. I guess I should put something here eventually */
	public void checkClientTrusted(X509Certificate[] chain, String authType) {
	}
 
	public void checkServerTrusted(X509Certificate[] chain, String authType) {
	}
 
	public X509Certificate[] getAcceptedIssuers() {
		return null;
	}
} 


