package bunshin;


import bunshin.util.*;
import bunshin.listeners.*;
import bunshin.storage.*;

import rice.environment.Environment;
import rice.p2p.commonapi.*;

import rice.pastry.NodeHandle;
import rice.pastry.NodeIdFactory;
import rice.pastry.PastryNode;
import rice.pastry.PastryNodeFactory;
import rice.pastry.commonapi.*;
import rice.pastry.dist.*;
import rice.pastry.socket.SocketPastryNodeFactory;
import rice.pastry.standard.*;
import rice.pastry.direct.*;

import java.io.*;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URL;
import java.net.UnknownHostException;
import java.util.*;
import java.nio.channels.*;

/**
 *
 * This class wraps a connection to Bunshin substrate and is used by the application as its main
 * DHT substrate.
 *
 * @author Ruben Mondejar  <Ruben.Mondejar@estudiants.urv.es>
 */

public class BunshinConnection {
  
  // Loads pastry settings
  public static Environment env = new Environment();

  // the port to begin creating nodes on
  public static int PORT = 5019;

  // the port on the bootstrap to contact
  public static int BOOTSTRAP_PORT = 5019;

  // the host to boot the first node off of
  public static String BOOTSTRAP_HOST = "localhost";

  // Reference to the Pastry node
  private PastryNode node;
  
  protected BunshinImpl bunshinApp;

  protected Endpoint endPoint;

  // ----- ATTEMPT TO LOAD LOCAL HOSTNAME -----
  static {
    try {
      BOOTSTRAP_HOST = InetAddress.getLocalHost().getHostName();
    } catch (UnknownHostException e) {
    }
  }

  /**
   * Creates a node and stablishs the connection with the network
   * @param String host
   * @param int port   
   *
   */
  public BunshinConnection(String bootHost, int bootPort) {
    createNode(bootHost,bootPort);
  }

  /**
   * Creates a node and stablishs the connection with the network
   * @param Properties bunshin file
   */
  public BunshinConnection(Properties prop) throws Exception {

      String bootHost =(String)prop.get(Context.HOST);
      if (bootHost==null) bootHost = "auto";

      int bootPort = Integer.parseInt((String)prop.get(Context.PORT));

      createNode(bootHost,bootPort);
  }

  /**
   * Creates a node and stablishs the connection with the network
   * @param String of the path of the properties bunshin file
   */
  public BunshinConnection(String path) throws Exception {

    Properties prop = loadProp(path);

    String bootHost =(String)prop.get(Context.HOST);
    if (bootHost==null) bootHost = "auto";

    int bootPort = Integer.parseInt((String)prop.get(Context.PORT));

    createNode(bootHost,bootPort);
  }

  private Properties loadProp(String path) throws Exception {
  	FileInputStream  fis = new FileInputStream(path);
	Properties prop = new Properties();
	prop.load(fis);
	fis.close();
  	return prop;
  }

  /**
   * This method creates a new node which will serve as the underlying p2p layer
   *
   * @param args Constructor arguments: 0 - String bootstrap host; 1 - int bootstrap port; 2 - PastryNodeFactory factory; 3 - int protocol
   */
  public void createNode (String bootHost, int bootPort) {

    if (!bootHost.equalsIgnoreCase ("auto")) {
      BOOTSTRAP_HOST = bootHost;
    }
    BOOTSTRAP_PORT = bootPort;
    PORT = bootPort;
 
    //  Generate the NodeIds Randomly
    NodeIdFactory nidFactory = new RandomNodeIdFactory(env);

    PastryNodeFactory factory = null;
    //PastryNodeFactory factory = new DirectPastryNodeFactory (new RandomNodeIdFactory(env), new SphereNetwork(env), env);
 
      int newPort = PORT;

      // If working in remote mode, check if port is already bound 
        while (true) {
          DatagramChannel channel = null;
          try {
            channel = DatagramChannel.open();
            channel.configureBlocking (false);
            InetSocketAddress isa = new InetSocketAddress (newPort);
            channel.socket().bind (isa);
            channel.socket().close();
            channel.close();
      
//          construct the PastryNodeFactory, this is how we use rice.pastry.socket
            factory = new SocketPastryNodeFactory(nidFactory, newPort, env);
            break;
          } catch (Exception e) {
            newPort += 100;
            System.out.println ("Port " + (newPort - 100) + " already bound. Trying " + newPort + "...");
          }
          finally {
            channel.socket().close();
            try {
              channel.close();
            }
            catch (IOException ex) {

            }
          }
        }      
              
        // Get bootstrap reference
        InetSocketAddress address = new InetSocketAddress (BOOTSTRAP_HOST, BOOTSTRAP_PORT);     

        // This will return null if we there is no node at that location
        NodeHandle bootHandle = ((SocketPastryNodeFactory) factory).getNodeHandle(address);

        // construct a node, passing the null boothandle on the first loop will cause the node to start its own ring
        node = factory.newNode((NodeHandle) bootHandle);
        

    // The node may require sending several messages to fully boot into the ring
    while (!node.isReady()) {
      // Delay so don't busy-wait
      try {
        Thread.sleep (100);
      }
      catch (InterruptedException ex1) {
      }
    }
  } 
  
  /**
   * Initializes the Bunshin layer.
   *
   * @param String id of the application
   * @param StorageManager the object manager
   * @param int replicaFactor indicates the number of copies in the network
   * @param boolean activate cache mode
   * @param boolean activate debug mode
   */
  public void init(String id, StorageManager manager, int replicaFactor, boolean cache, boolean debug) {
  	bunshinApp = new BunshinImpl(id);
    bunshinApp.setStorageManager(manager);
    bunshinApp.setReplicationFactor(replicaFactor);
    if (cache) bunshinApp.activateCache();
    if (debug) bunshinApp.activateDebug();
    endPoint = node.registerApplication (bunshinApp,id);
    bunshinApp.setRefreshTime(20000);
    bunshinApp.setEndPoint (endPoint);
  }	
  	
    /**
   * Initializes the Bunshin layer.
   *
   * @param String id of the application
   * @param StorageManager the object manager
   * @param int replicaFactor indicates the number of copies in the network
   * @param int refresh time period in seconds 
   * @param boolean activate cache mode
   * @param boolean activate debug mode
   */	  	
  public void init(String id, StorageManager manager, int replicaFactor, int time_sec, boolean cache, boolean debug) { 	
    bunshinApp = new BunshinImpl(id);
    bunshinApp.setStorageManager(manager);
    bunshinApp.setReplicationFactor(replicaFactor);
    if (cache) bunshinApp.activateCache();
    if (debug) bunshinApp.activateDebug();
    endPoint = node.registerApplication (bunshinApp,id);
    bunshinApp.setRefreshTime(time_sec*1000);
    bunshinApp.setEndPoint (endPoint);
  }

  /**
   * Initializes the Bunshin layer.
   *
   * @param String path of the bunshin properties file
   */
  public void init(String path) throws Exception{

  	Properties prop = loadProp(path);

  	String id =(String)prop.get(Context.ID_APPLICATION);
  	StorageManager manager = (StorageManager)Class.forName((String)prop.get(Context.STORAGE_MANAGER)).newInstance();
  	manager.init(prop);
  	int replicaFactor = Integer.parseInt((String)prop.get(Context.REPLICA_FACTOR));
  	boolean cache = Context.TRUE.equals((String)prop.get(Context.CACHE));
  	boolean debug = Context.TRUE.equals((String)prop.get(Context.DEBUG));

    int time = Integer.parseInt((String)prop.get(Context.REFRESH_TIME));
  	init(id,manager,replicaFactor,time,cache,debug);
  	
  }
  /**
  * Obtains the Bunshin's application reference.
  * @return bunshinApp
  */
  public Bunshin getBunshinApp() {
	  return bunshinApp;
  }
  
  /**
   * Sets the information of a specific context (optional method) 
   * @param context Id of the context
   * @param path String path of the storage directory (only DiskStorage) 
   * @param url String url base (ex : http://www.infourl.com)
   * @param extra paths of class/jar for classLoader
   */
  public void setInfoContext(String context, String path, String url, URL[] URLsList) {
    bunshinApp.setInfoContext(context,path,url,URLsList);
  }
  
  /**
   * Asynchronous get method
   * @param String value
   * @param BunshinGetClient the callback client
   */
  public void retrieveObject(String context, Id key, bunshin.listeners.BunshinGetClient client) throws IOException {
  	bunshinApp.get(context,key,"Object", client);
  }


  /**
   * Puts the pair (key,value)
   * @param Id of the key
   * @param Serializable value
   */
  public void storeObject(Id id, Serializable value) {
    bunshinApp.put(id,value,"Object");
  }
  
  /**
   * Puts the pair (key,value)
   * @param Id of the key
   * @param Serializable value
   * @param String field
   */
  public void storeObject(Id id, Serializable value, String field) {
    bunshinApp.put(id,value,field);
  }

	/**
	 * Puts the pair (key,value)
	 * @param Context string
	 * @param Id of the key
	 * @param Serializable value
	 */
	public void storeObject(String context,Id id, Serializable value)
	{
		bunshinApp.put(context,id,value,"Object");
	}

	/**
	 * Puts the pair (key,url), when value = download(url);
	 * @param Id of the key
	 * @param URL url of the value
	 */
	public void insertURL(String context,Id id, URL url) {
	  bunshinApp.putURL(context,id,url,"URL");
	}

  /**
   * Puts the pair (key,value), when key = value
   * @param String of the key
   * @param Serializable value
   * @return Id of the key
   */
  public Id storeObject(String key, Serializable value) {
  	Id id = Utilities.generateHash(key);
    bunshinApp.put(id,value,"Object");
    return id;
  }

  /**
   * Synchronous get method
   * @param String of the key
   * @return Object value of this id key
   */
  public Object retrieveObject(String key) throws IOException {
  	Id id = Utilities.generateHash(key);
  	return retrieveObject(id);
  }
  
  /**
   * Synchronous get method
   * @param Id of the key
   * @return Object value of this id key
   */
  public Object retrieveObject(String context, Id key) throws IOException {
    return retrieveObject(context,key,"Object");
  }

  /**
   * Synchronous get method
   * @param Id of the key
   * @return Object value of this id key
   */
  public Object retrieveObject(Id key) throws IOException {
    return retrieveObject(Context.DEFAULT_CONTEXT,key,"Object");
  }
  
  /**
   * Synchronous get method
   * @param Id of the key
   * @param String field
   * @return Object value of this id key
   */
  public Object retrieveObject(Id key, String field) throws IOException {
    return retrieveObject(Context.DEFAULT_CONTEXT,key,field);
  }

  private class BunshinGetClient implements bunshin.listeners.BunshinGetClient
  {
    public boolean lookupResultArrived = false;
    public Object value = null;
    public String field;

    public BunshinGetClient(String _field)
    {
        field = _field;
    }

	public void get (Object result) {

	    synchronized (this) {
	        if (result != null) {

                if (result instanceof bunshin.util.Values)
                {
                    value =((bunshin.util.Values) result).get(field);
                }
                else
                {
                    value = result;
                }
            }
            else
            {
                value = null;
            }
            lookupResultArrived = true;
        }
    }
  }

	/**
	 * Synchronous get method
	 * @param Context String
	 * @param Id of the key
	 * @return Object value of this id key
	 */
	public Object retrieveObject(String context,Id key, final String field) throws IOException {

        System.err.println("RetrieveObject key: " + key.toStringFull());
		int timeout = 0;
        final Object value = null;

        BunshinGetClient client = new BunshinGetClient(field);

		bunshinApp.get(context,key,field, client);

		synchronized (client)
		{
			while (!client.lookupResultArrived && timeout < 30)
			{
				try
				{
					client.wait (100);
				}
				catch (InterruptedException ex)
				{
				}
				timeout++;
			}
		}

        System.err.println("RetrieveObject key " + key.toStringFull() + " returning value: " + value  + " timeout " + timeout);
		return client.value;
	}

  private class BunshinURLClient implements bunshin.listeners.BunshinURLClient
  {
    public boolean lookupResultArrived = false;
    public URL urlValue = null;

    public BunshinURLClient()
    {
    }

	public void get (URL result) {

	    synchronized (this) {
	        if (result != null) {
                    urlValue = result;
            }
            else
            {
                urlValue = null;
            }
            lookupResultArrived = true;
        }
    }
  }
		/**
	 * Synchronous get method
	 * @param Context String
	 * @param Id of the key
	 * @return Object value of this id key
	 */
	public URL retrieveURL(String context,Id key) throws IOException {

		int timeout = 0;

        BunshinURLClient client = new BunshinURLClient();

		bunshinApp.getURL(context,key,"URL",client);

		synchronized (client)
		{
			while (!client.lookupResultArrived && timeout < 30)
			{
				try
				{
					client.wait(100);
				}
				catch (InterruptedException ex)
				{
				}
				timeout++;
			}
		}

		return client.urlValue;
	}


   	/**
	 * Synchronous get method
	 * @param Context String
	 * @param Id of the key
	 * @return Object value of this id key
	 */
	public Object retrieve(String context,Id key, final String field) throws IOException {

		int timeout = 0;

        BunshinGetClient client = new BunshinGetClient(field);

		bunshinApp.get(context,key,field,client);

		synchronized (client)
		{
			while (!client.lookupResultArrived && timeout < 30)
			{
				try
				{
					client.wait (100);
				}
				catch (InterruptedException ex)
				{
				}
				timeout++;
			}
		}

		return client.value;
	}




  /**
   * Remove method
   * @param String of the key
   */
  public void removeObject(String key) throws IOException
  {
  	Id id = Utilities.generateHash(key);
  	bunshinApp.remove(id,"Object");
  }

  /**
   * Remove method
   * @param Id of the key
   */
  public void removeObject(Id idValue) throws IOException {
  	bunshinApp.remove(idValue,"Object");
  }
  
  /**
   * Remove method
   * @param String key
   * @param String field
   * @throws IOException
   */
  public void removeObject(String key, String field) throws IOException
  {
  	Id id = Utilities.generateHash(key);
  	bunshinApp.remove(id,field);
  }

  /**
   * Remove method
   * @param Id idValue
   * @param String field
   * @throws IOException
   */
  public void removeObject(Id idValue, String field) throws IOException {
  	bunshinApp.remove(idValue,field);
  }

  /**
   * Leaves the bunshin layer
   */
  public void leave() {
  	bunshinApp.leave();
  }


}
