/* ***************************************************************************+
 * ITX package (cnrg.itx) for telephony application programming.              *
 * Copyright (c) 1999  Cornell University, Ithaca NY.                         *
 * A copy of the license is distributed with this package.  Look in the docs  *
 * directory, filename GPL.  Contact information: bergmark@cs.cornell.edu     *
 ******************************************************************************/


package cnrg.itx.datax;

import java.util.*;
import java.net.*;
import java.io.*;
import cnrg.itx.datax.devices.*;

/**
 * Class representing a channel in the <code>Connection</code> object. A <code>Channel</code>
 * acts as a pipe between a <code>Source</code> and all the <code>Destination</code> objects
 * and copies data between them.
 */
public class Channel implements Runnable, Statistics, Properties
{
	/**
	 * Default channel sample size
	 */
	public static final int SAMPLE_SIZE = 64;
	
	/**
	 * Constant to specify that a channel is an input channel.
	 */
	public static final int INPUT = 0;
	
	/**
	 * Constant to specify that a channel is an output channel.
	 */
	public static final int OUTPUT = 1;
	
	/**
	 * Constant to specify that a channel is neither an output nor input channel (maybe a hybrid).
	 */
	public static final int OTHER = 2;
	
	/**
	 * Attribute to hold the mode of the channel
	 */
	protected int mode = OTHER;
	
	/**
	 * Attribute to store the source object.
	 */
	protected Source source;
	
	/**
	 * Attribute to store the list of destination objects.
	 */
	protected Vector destinations;
	
	/**
	 * Flag for specifying if the channel is running.
	 */
	protected boolean running;
	
	/**
	 * Flag for specifying if the channel has been closed
	 */
	protected boolean closed;
	
	/**
	 * Input pipe
	 */
	protected PipedInputStream pipedInput;
	protected PipedOutputStream pipedOutput;
	
	/**
	 * Reader thread
	 */
	protected Thread readerThread;
	
	/**
	 * Sample size
	 */
	protected int sampleSize;
	
	/**
	 * Default constructor for a channel. Constructs a null channel.
	 */
	public Channel()
	{	
		this(SAMPLE_SIZE);
	}	
	/**
	 * Constructor for a channel. Constructs a null channel.
	 * 
	 * @param sampleSize size (in bytes) read by the channel read thread
	 */
	public Channel(int sampleSize)
	{	
		// Initialize the destination vector
		destinations = new Vector();
		
		// Initialize flags
		running = false;
		closed = false;
		
		// Sample size
		this.sampleSize = sampleSize;
	}

	/**
	 * Method to set the source for the Channel.
	 * @param source The Source for the Channel
	 */
	public void setSource(Source source)
	{
		this.source = source;
	}
	
	/**
	 * Method to get the source for the Channel.
	 * @return Source The Channel's Source object
	 */
	public Source getSource()
	{
		return source;
	}
	
	/**
	 * Method to set the mode for the Channel.
	 * @param mode The new mode
	 */
	private void setMode(int mode)
	{
		this.mode = mode;
	}
	
	/**
	 * Method to get the mode for the channel.
	 * @return mode the current mode of the channel
	 */
	private int getMode(int mode)
	{
		return mode;
	}
	
	/**
	 * Method to add a destination for the Channel. This will add the destination to the
	 * list of destinations.
	 * @param d The destination to add to the list of destination for the channel
	 * @exception DuplicateDestinationException
	 */
	public void addDestination(Destination d) throws DuplicateDestinationException
	{
		// Check if the destination is alrady present in the list
		if(destinations.contains(d))
		{
			throw new DuplicateDestinationException("Destination already present in Channnel");
		}

		// Add the destination to the list
		destinations.addElement(d);
	}
	
	/**
	 * Method to open the Channel. This method starts the channel thread that copies data
	 * from the source to all the destinations.
	 */
	public void open() throws DataException
	{
		if (closed)
		{
			throw new DataException("Channel closed");
		}
		
		// Create the input pipe
		try
		{
			pipedOutput = new PipedOutputStream();
			pipedInput = new PipedInputStream();
			pipedOutput.connect(pipedInput);		
		}
		catch (IOException e)
		{
			throw new DataException(e.getMessage());
		}
		
		// Create and start the reader thread
		running = true;
		readerThread = new Thread(this);
		readerThread.start();
		
		// Start the source
		source.start();
	}
	
	/**
	 * Channel reader thread function.
	 */
	public void run()
	{
		byte[] data = new byte[sampleSize];
		int length;
		int numRead;
		int totalRead;
		
		while (running)
		{
			try
			{
				// Read sampleSize bytes from the input pipe
				totalRead = 0;
				length = sampleSize;
				
				while (totalRead < sampleSize)
				{
					numRead = pipedInput.read(data, totalRead, length);
					
					// Check for EOF
					if (numRead == -1)
					{
						throw new IOException();
					}
					
					totalRead += numRead;
					length -= numRead;
				}
				
				if (running)
				{
					for (Enumeration e = destinations.elements(); e.hasMoreElements() ; )
					{
						((Destination)e.nextElement()).write(data);
					}
					
				}
			}
			catch (IOException e)
			{
				// Pipe has been broken, so quit...
				break;
			}
			catch (DataException de)
			{
				// FIXME: what to do with exceptions?
				de.printStackTrace();
			}			
		}
	}
	
	/**
	 * Method that receives the data from the source.  
	 * <p> Note: the buffer that is passed should persist for some time.  
	 * @param bData buffer that will be guaranteed to be unchanged for some amount of time.
	 * FIXME: what to do with excetions?
	 */
	public void push(byte [] bData)
	{
		try
		{
			pipedOutput.write(bData);
		}
		catch (IOException e)
		{
			// FIXME: what to do with exceptions?
			e.printStackTrace();
		}
	}

	/**
	 * Method to close the Channel. This method first stops the channel thread and then
	 * clears all the resources held by the Channel.
	 */
	public void close() 
	{
		// Return if the channel is closed
		if (closed)
		{
			return;
		}
		
		// Clear all the resources. Close the source and all the destinations
		source.close();

		// Stop the reader thread
		if (running)
		{
			running = false;
			try
			{
				pipedInput.close();
				pipedOutput.close();
			}
			catch (IOException e)
			{
			}
		
			// Wait for the thread to die
			try
			{
				readerThread.join(100);
			}
			catch (InterruptedException e)
			{
			}
		}
		
		// Close all destinations
		for(Enumeration e = destinations.elements() ; e.hasMoreElements();)
		{
			((Destination)e.nextElement()).close();
		}
		
		closed = true;
	}
	
	/**
	 * Method to remove a destination from the list.
	 * @param d The Destination object to remove
	 * @return boolean True if the destination was found and removed
	 */
	public boolean removeDestination(Destination d)
	{
		return destinations.removeElement(d);
	}
	
	/**
	 * Method to remove all destinations.
	 */
	public void removeAllDestinations()
	{
		destinations.removeAllElements();
	}
	
	/**
	 * Method to mute all the sources and destinations
	 * @param state The state of the mute. true to mute and false to unmute
	 */
	public void mute(boolean state)
	{
		// Mute all the resources.
		source.mute(state);
		
		for(Enumeration e = destinations.elements() ; e.hasMoreElements();)
		{
			((Destination)e.nextElement()).mute(state);
		}
	}

	/**
	 * Method to get the statistics from the channel. This method gets the
	 * statistics from the devices connected to the channel.
	 * @return Stats the statistics for the channel
	 */
	public Stats getStatistics ()
	{
		Stats s = new Stats();
		for (Enumeration e = destinations.elements() ; e.hasMoreElements() ; )
		{
			s.merge( ((Destination)e.nextElement()).getStatistics() );
		}
		
		if (source != null)
			s.merge(source.getStatistics());
		
		return s;
	}

	/**
	 * Gets the properties for this audioConnection
	 * @return the ProertiesCollection
	 */
	public PropertiesCollection getProperties()
	{
		PropertiesCollection pc = null;
		
		try
		{
			pc = source.getProperties();
			for (Enumeration e = destinations.elements() ; e.hasMoreElements() ; )
			{

				if (pc != null)
					pc.merge( ((Destination)e.nextElement()).getProperties());
				else
					pc =  ((Destination)e.nextElement()).getProperties();
			}
		}
		catch (DataException e)
		{
			e.printStackTrace();
		}
		
		return pc;
	}

	/**
	 * Sets the properties for this audioConnection
	 */
	public void setProperties(PropertiesCollection pc)
	{
		try
		{
			source.setProperties(pc);
			for (Enumeration e = destinations.elements() ; e.hasMoreElements() ;)
			{
				((Destination)e.nextElement()).setProperties(pc);
			}
		}
		catch (DataException e)
		{
			e.printStackTrace();
		}
	}

	/**
	 * Interface to set the given properties collection into the device. WOrks under the 
	 * assumption that this is the properties collection of the peer.
	 */
	public void setPeerProperties(PropertiesCollection pc) throws DataException
	{
		try
		{
			if (source != null)
				source.setPeerProperties(pc);

			for (Enumeration e = destinations.elements() ; e.hasMoreElements() ;)
			{
				((Destination)e.nextElement()).setPeerProperties(pc);
			}
		}
		catch (DataException e)
		{
			e.printStackTrace();
		}
	}
}