package edu.uprm.admg.nettraveler.schema;
// JDK imports
import java.io.Externalizable;
import java.io.IOException;
import java.io.InvalidObjectException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import edu.uprm.admg.nettraveler.system.NetTravelerException;
import edu.uprm.admg.nettraveler.type.MDObject;
import edu.uprm.admg.nettraveler.type.MDSmallObject;
import edu.uprm.admg.util.DataVerify;

/**
 * <code>Tuple</code> represents a tuple that is being processed
 * in the execution engine. Each tuple contains
 * the values of one or more attributes, represented as specific 
 * instances of sub-classes from the class <code>MDObject</code>. 
 * <p/>
 * Each tuple is created from a template that consists of an array
 * of attribute descriptions (instances of class <code>Attribute</code>).
 * Individual attribute values (columns) can be accessed based on 
 * their position in the tuple. Moreover, each tuple can be serialized
 * to an object output stream, and de-serialized from an object input
 * stream.
 * 
 * @author M.Rodriguez-Martinez
 * @author Elliot A. Vargas-Figueroa
 */

public final class Tuple implements Externalizable {
    /**
	 * Serial Version
	 */
	private static final long serialVersionUID = 4199905870674039914L;
	/**
	 * Class Version
	 */
	private final static int VERSION = 1;
	/**
	 * Cached hash code
	 */
	private transient int cachedHash = 0;
	/** 
     * The values for each of the attributes (columns) in
     * this tuple.
     */
    private transient List<MDObject> columns = null;
    /**
     * Name of the data source from which this tuple originally came.
     */
    private transient String sourceAddress = null;
    /**
     * Constructs a new <code>Tuple</code> with no columns.
     * <p/>
     * This constructor is here with the sole purpose of instantiating
     * a <code>Tuple</code> before reading
     *  its contents from a stream using
     * the {@link #readExternal(ObjectInput)} method.
     */
    public Tuple(){}
    
    /** 
     * Constructs a new <code>Tuple</code> from an array of attribute
     * objects that serves as the template for the stucture of each of 
     * the attribute values to be stored in the tuple.
     * @param template an array with the information about the attributes present in the tuple.
     * @exception TupleInstantiationException if an error occurs while initializing the structures used to hold the attribute values.
     */
    public Tuple(Attribute[] template)throws TupleInstantiationException{
	    	// sanity check
	    	DataVerify.VerifyArrayParam("template", template);
	    	this.columns = new ArrayList<MDObject>(template.length);
	    	sourceAddress = null;
	    	MDObject o;
	    	for (int i=0; i < template.length; ++i){
	    		try {
	    			o = (MDObject) template[i].getBaseClass().newInstance();
	    			/* Set any constant values now */
	    			if(template[i] instanceof ValueAttribute){
	    				if(!(template[i] instanceof FunctionValueAttribute))
	    					o.resetValue(((ValueAttribute) template[i]).getValue());
	    			}
	    			columns.add(o);
	    		}
	    		catch(IllegalAccessException e){
	    			throw new 
	    			TupleInstantiationException("Unable to access the" +
	    					"definition of class " + 
	    					template[i].getBaseClass().getName() + ".");
	    		}
	    		catch(InstantiationException e){
	    			throw new 
	    			TupleInstantiationException
	    			("Unable to instantiate class " + template[i].getBaseClass().getName() + 
	    			"because it is either abstract or an interface.");
	    		}
	    	}
    }
    /**
     * Returns the value of an attribute based on its position in the 
     * tuple. The range of valid positions is <code>[0, N-1]</code>, where
     * <code>N</code> is the number of attributes (columns) in the 
     * tuple.
     * </P>
     * <i>Caution</i>: This method assumes that the position value passed
     * as parameter is within the appropriate range. Otherwise, an runtime 
     * exception will occur.
     * 
     * @param position the position of the attribute in the tuple.
     * @return the value of the attribute stored at the given position.
     */
    public MDObject getAttribute(int position) {
    		DataVerify.VerifyIntegerRange("position", position, 0, getNumAttributes() - 1);
	    	return (MDObject) columns.get(position);
    }
    /**
     * Sets the attribute value in the given position in the array.
     * @param position Position of the value to change
     * @param new_value New value.
     */
    public void setAttributeValue(int position, MDObject new_value){
		DataVerify.VerifyIntegerRange("position", position, 0, getNumAttributes() - 1);
		((MDObject) columns.get(position)).resetValue(new_value);
    }
    /**
     * This method will drop off from this Tuple 
     * all columns whose indexes are not
     * in <code>projs</code>.
     * <p/>
     * The passed array must containt the indexes of the 
     * attributes to be preserved by the <code>Tuple</code>.
     * 
     * @param projs Indexes of the attributes to project.
     */
    public void keepColumns(int[] projs){
    		DataVerify.VerifyObjectParam("projs", projs);
    		List<MDObject> newColumns = new ArrayList<MDObject>(projs.length);
    		//get projected columns
    		for(int i=0; i<projs.length; i++)
    			newColumns.add(columns.get(projs[i]));
    		//drop the other columns
    		columns.clear();
    		columns = newColumns;
    		newColumns = null;
    }
    /**
     * Returns the number of attributes (columns) in this tuple. 
     * @return number of attributes in this tuple.
     */
    public int getNumAttributes() {
    		return columns.size();
    }
	/**
	 * Returns the address of data source
	 * for this tuple. 
	 * 
	 * @return Returns the igName.
	 */
	public String getSourceAddress() {
		return sourceAddress;
	}
	/**
	 * Returns the address of data source
	 * for this tuple.
	 * 
	 * @param addr The address to set.
	 */
	public void setSourceAddress(String addr) {
		this.sourceAddress = addr;
	}
    /**
     * Returns the size in bytes of this tuple.
     * @return the size in bytes of this tuple.
     */
	public int getSize(){
		int result = 0;
		for (int i=0; i < getNumAttributes(); ++i)
			result += getAttribute(i).size();
		
		return result;
	}
    /**
     * Returns a string representation of this <code>Tuple</code>
     */
    public String toString(){
    		StringBuffer buffer = new StringBuffer();
    		for (int i=0; i < getNumAttributes(); ++i)
    			buffer.append(getAttribute(i).toString() + " ");
    		return buffer.toString();
    }
    
    /**
     * Return <code>true</code> if this Tuples is equal to 
     * the passed Tuple.
     * </p>
     * Columns that are not of type 
     * {@link edu.uprm.admg.nettraveler.type.MDSmallObject} are 
     * not considered during the comparison. 
     * 
     * @param tuple The <code>Tuple</code> to compare with.
     * @return <code>true</code> if this <code>Tuple</code> is
     * 		equal to the passed Tuple.
     */
    public boolean equals(Object tuple){
	    	if(this == tuple)
	    		return true;
	    	if(!(tuple instanceof Tuple))
	    		return false;
	    	Tuple t = (Tuple) tuple;
	    	if(t.getNumAttributes() != this.getNumAttributes())
	    		return false;
    		MDObject o1;
    		MDObject o2;
	    	for(int i=0; i<this.getNumAttributes(); i++){
	    		o1 = this.getAttribute(i);
	    		o2 = t.getAttribute(i);
	    		if((o1 instanceof MDSmallObject ) && (o2 instanceof MDSmallObject))
	    			if(!(o1.equals(o2)))
	    				return false;
	    	}
	    	return true;
    }
    /**
     * Returns the hash code.
     * 
     * @return This <code>Tuple</code> hash code.
     */
    public int hashCode(){
	    	if(this.cachedHash == 0){
	    		this.cachedHash = this.calculateHashCode();
	    	}
	    	return this.cachedHash;
    }
    /**
     * This method is the one that calculates this object
     * hash code.
     * @return hash code.
     */
    private int calculateHashCode(){
	    	int result = 17;
	    	MDObject o;
	    	for(int i=0; i<getNumAttributes(); i++){
	    		o = getAttribute(i);
	    		if(o instanceof MDSmallObject)
	    			result += 37 * result + o.hashCode();
	    	}
	    	return result;
    }
	/**
	 * This method returns a <code>List</code> of <code>Tuple</code>
	 * objects constructed with the specified <code>template</code>.
	 * </p>
	 * The size of the <code>List</code> is specified by
	 * <code>size</code>.
	 * 
	 * @param template Template for the <code>Tuples</code>.
	 * @param size Size of the returned <code>List</code>.
	 * @return <code>List</code> of tuples.
	 * @throws NetTravelerException If any error occurs while generating the 
	 * <code>List</code>.
	 */
	public static List getListOfTuples(Attribute[] template, int size)
				throws NetTravelerException{
        Tuple[] t = new Tuple[size];
        try{
	        for(int i=0; i<t.length; i++){
	            t[i] = new Tuple(template);
	        }
        }catch(Exception e){
        		throw new NetTravelerException(e);        	
        }
        return Arrays.asList(t);	
	}
    /**
     * Writes the contents of this tuple to a non-null object output 
     * stream. The tuple to be written cannot be one that have been
     * constucted with the empty constructor.
     * 
     * @param out an object output stream to write the columns in the tuple. The output stream cannot be null.
     * @exception IOException if an I/O error occurs while writting the data.
     */
    public void writeExternal(ObjectOutput out) throws IOException {
    		if(columns == null)
    			throw new NullPointerException("Attributes of the Tuple are null");
//    		if(sourceAddress == null)
//    			throw new NullPointerException("Address of the Tuple's source cannot be null");
	    	//Write this class version
	    	out.writeInt(VERSION);
	    	// first write the number of columns
	    	int num_columns = getNumAttributes();
	    	out.writeInt(num_columns);
	    	/*
	    	 * Now write the contents of each column.
	    	 * Since I want to be able to deserialized this
	    	 * Tuple even if the receiving Tuple object has
	    	 * not been initialize I will send the Type
	    	 * of the Object for each of the columns .
	    	 */
    		MDObject o;
	    	for (int i=0; i < num_columns; ++i){
	    		o = getAttribute(i);
	    		out.writeUTF(o.getClassName());
	    		o.writeExternal(out);
	    	}
    		if(sourceAddress == null)
    			/* Write a dummy address */
    			out.writeUTF("");
    		else 
    			//Write the data source address
    			out.writeUTF(this.sourceAddress);
    }    
    /**
     * Reads the content to be assigned to this tuple from a non-null object
     * input stream.
     * <i>Caution</i>: This method assumes that the tuple has been initialized
     * with the appropriate constructor indicating the column data types,
     * in order to properly read the column values. Also, it assumes that 
     * the content for the tuple was written using the 
     * <code>writeExternal</code> method defined in the class.
     *  
     * @param in an object input stream to read the columns for this tuple. The input stream cannot be null.
     * @exception ClassNotFoundException if the class for one of the columns in the tuple cannot be found.
     * @exception IOException if an I/O error occurs while reading the tuple.
     */
    public void readExternal(ObjectInput in) throws ClassNotFoundException, IOException{
	    	/*
	    	 * Read the object version. This object will do nothing
	    	 * with this number since this is the first version
	    	 * of this class. Subsequent versions of
	    	 * this class MUST work with the version number.
	    	 */
	    	in.readInt();
	    	//Read the number of columns
	    	int temp = in.readInt();
	    	if (temp <= 0)
	    		throw new InvalidObjectException("Number of columns in Tuple " +
	    				"must be greater than 0");
	    	boolean initialize = false;
	    	//Check if the Tuple was previously initialized
	    	if(this.columns != null){
	    		/*
	    		 * Check that the received number of columns equals
	    		 * this Tuple number of columns
	    		 */
	    		if(temp != this.getNumAttributes())
	    			throw new InvalidObjectException("Received number of columns " + 
	    					temp + " did not matched the" +
	    					" expected number of columns " + getNumAttributes());
	    	}else{
	    		// This Tuple was not initialized so
	    		 // I must initialize all columns
	    		 // before reading them
	    		 this.columns = new ArrayList<MDObject>(temp);
	    		 initialize = true;
	    	}
	    	String cName;
	    	MDObject o;
	    	for(int i=0; i<temp; i++){
	    		cName = in.readUTF();
	    		/*
	    		 * Initialize each column if the Tuple
	    		 * was not previously initalized
	    		 */
	    		if(initialize){
	        		try {
	        			this.columns.add((MDObject)Class.forName(cName).newInstance());
	        		}
	        		catch(IllegalAccessException e){
	        			throw new 
	        			IOException("Unable to access the" +
	        					"definition of class " + 
	        					cName);
	        		}
	        		catch(InstantiationException e){
	        			throw new 
	        			IOException
	        			("Unable to instantiate class " + cName + 
	        			"because it is either abstract or an interface");
	        		}
	    		}
	    		o = getAttribute(i);
	    		o.readExternal(in);
	    	}
	    	this.sourceAddress = in.readUTF();
	    	/* Clean */
	    	cName = null;
    }
}