/* ***************************************************************************+
 * 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 cnrg.itx.datax.devices.*;
import java.util.*;
import java.lang.Math;
import java.io.*;
/**
 * MixerChannel acts as a channel with multiple inputs. Destinations can be set (as in a standard
 * Channel) with the addDestination() method. To use multiple inputs, call the getNewInput method.
 * This returns a Channel object that can be assigned a source. Example: 
 * <p>
 * The following creates a MixerChannel and adds a microphone soruce and a mixer source to it.
 * <pre> 
 * Channel ch;
 * MixerChannel mc = new MixerChannel();
 * 
 * ch = mc.getNewInput();
 * ch.setSource(new MicrophoneSource(ch));
 * ch = mc.getNewInput();
 * ch.setSource(new NetworkSource(...,ch,...));
 * </pre>
 */
public class MixerChannel extends Channel implements Runnable{
	
	
	/**
	 * a vector containing the input ChannelBuffers to feed into the mixer.
	 */
	private Vector inChannels;
	
	/**
	 * A post-mixing buffering channel. This stage is enabled by the useOutputChannel variable.
	 */
	private Channel outChannel;
	/**
	 * number of bytes the mixerChannel has sent so far
	 */
	private long outBytes = 0; 
	/**
	 * The number of bytes to be mixed at a time. This is also the number of bytes we send at a time to our destination.
	 */
	private int mixSampleSize = Channel.SAMPLE_SIZE; 
	/*
				private FileOutputStream fos;
				private PrintWriter pw;
	*/
				
				
	/**
	 * The number of ms we try to hold buffered at our destination.
	 */
	private int BUFFER_TIME= 100;
				
	/**
	 * The sample rate of the data to be mixed.
	 */
	private int SAMPLE_RATE = MicrophoneSource.SAMPLE_RATE;
	
	/**
	 * The number of time we have to process each byte in ms.
	 */
	private float SAMPLE_TIME = (float)1000/SAMPLE_RATE;
	
	/**
	 * Determines if we will use a second buffering stage after the mixer. 
	 * If this is true, we will allocate a sepatate channel with the mixer as
	 * it's source. If this is false, things work marginally faster (no extra buffering)
	 * but we have to set mixSampleSize to the destination's sample size. 
	 */
	private boolean useOutputChannel = false;
	/**
	 * Number of instances of the mixer; used for statistics output numbering.
	 */
	static int instances = 0;
	
	/**
	 * Instance number of the current mixer; used for stat output numbering.
	 */
	private int iMyInstance;
	
	/**
	 * Total bytes of silence mixed in to samples we've sent.
	 */
	private int totalSilence;
	
	/**
	 * holds correspondence between linked inputs and outputs for source muting in conferences.
	 */
	private Hashtable ioLinks = null;

	/**
	 * Creates an mixerChannel with no sources or destinations.
	 */
	public MixerChannel () {
		super();
		doSetup();
		
	}
	
	/**
	 * Creates a new mixerChannel with the specified sample size.
	 * @param sampleSize the number of bytes to be sent by the channel in each write.
	 */
	public MixerChannel(int sampleSize) {
		super(sampleSize);
		doSetup();
	}
	
	/**
	 * Setup steps common to both constructors.
	 */
	private void doSetup() {
		iMyInstance = instances++;
		inChannels = new Vector();
		ioLinks = new Hashtable();
		if (useOutputChannel) {
			outChannel = new Channel(Channel.SAMPLE_SIZE);
			outChannel.setSource(new DummySource());
		} else {
			this.setSource(new DummySource());
		}
		/* Debug...
		try{
			fos = new FileOutputStream("mixer.pcm");
			
			pw = new PrintWriter(fos);
		} catch(Exception e){if (fos == null) System.exit(24);}
		*/
	}

	/**
	 * 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 thrown when d is already a destination of this channel.
	 */
	public void addDestination(Destination d) throws DuplicateDestinationException {
		if (useOutputChannel) 
			outChannel.addDestination(d);
		else 
			super.addDestination(d);
	}
	
	/**
	 * 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){
		if (useOutputChannel)
			return outChannel.removeDestination(d);
		else
			return super.removeDestination(d);
	}
	
	/**
	 * Method to remove all destinations.
	 */
	public void removeAllDestinations() {
		if (useOutputChannel)
			outChannel.removeAllDestinations();
		else 
			super.removeAllDestinations();
	}
	
	/**
	 * Method to open the Channel. This method starts the mixer thread that mixes data
	 * from the sources and sends it to all the destinations. It also starts all of the
	 * mixer's sources and destinations.
	 * @exception DataException thrown when channel, or components of channel have already been closed.
	 */
	public void open() throws DataException {
		
		if (closed)
		{
			throw new DataException("Channel closed");
		}
		
		if (running) {
			return;
		}
		
		
		//start the output buffer
		if (useOutputChannel)
			outChannel.open();

		//start the mixer thread
		running = true;
		readerThread = new Thread(this);
		//readerThread.setPriority(Thread.MAX_PRIORITY);
		readerThread.setName("Mixer Thread");
		readerThread.start();
		
		// Start the sources
		for (Enumeration e = inChannels.elements(); e.hasMoreElements() ; ){
			((ChannelBuffer)e.nextElement()).open();
		}
	}
	
	/**
	 * 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) {
		if(useOutputChannel) 
		   outChannel.mute(state);
		else {
			for (Enumeration e = destinations.elements(); e.hasMoreElements(); )
				((Destination)e.nextElement()).mute(state);
		}
		for (Enumeration e = inChannels.elements(); e.hasMoreElements() ; ) {
			((ChannelBuffer)e.nextElement()).mute(state);
		}
	}
	
	/**
	 * Gets the properties for this MixerChannel
	 * @return the ProertiesCollection
	 */
	public PropertiesCollection getProperties() {
		
		PropertiesCollection pc = null;
		
		//try{
			
			if (useOutputChannel)
				pc = outChannel.getProperties();
			else 
				pc = super.getProperties();
			/*
			for (Enumeration e = inChannels.elements() ; e.hasMoreElements() ; ){
			
				if (pc != null)
					pc.merge( ((ChannelBuffer)e.nextElement()).getProperties());
				else
					pc =  ((ChannelBuffer)e.nextElement()).getProperties();
			}
			*/
		//}catch (DataException e){
		//	e.printStackTrace();
		//}
		
		return pc;
	}
	
	/**
	 * Sets the properties for this MixerChannel
	 * @param pc The new PropertiesCollection for this MixerChannel
	 */
	public void setProperties(PropertiesCollection pc) {
		if (useOutputChannel)
			outChannel.setProperties(pc);
		else {
			try {
				for (Enumeration e = destinations.elements() ; e.hasMoreElements() ;){
					((Destination)e.nextElement()).setProperties(pc);
				}
			}catch (DataException de) {
				de.printStackTrace();
			}
		}
		for (Enumeration e = inChannels.elements() ; e.hasMoreElements() ;){
			((ChannelBuffer)e.nextElement()).setProperties(pc);
		}
	}
	
	/**
	 * 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 (useOutputChannel)
				outChannel.setPeerProperties(pc);
			else {
				for (Enumeration e = destinations.elements() ; e.hasMoreElements() ;){
					((Destination)e.nextElement()).setPeerProperties(pc);
				}
			}

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

	/**
	 * 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();
		s.addStat("Device", "Mixer");
		s.addStat("<mixer "+iMyInstance+"> Bytes sent",new Integer((int)outBytes));
		s.addStat("<mixer "+iMyInstance+"> Silent bytes mixed",new Integer((int)totalSilence));
		s.addStat("<mixer "+iMyInstance+"> Number of sources",new Integer(inChannels.size()));
		if (useOutputChannel)
			s.merge(outChannel.getStatistics());
		else {
			for (Enumeration e = destinations.elements() ; e.hasMoreElements() ;){
				s.merge(((Destination)e.nextElement()).getStatistics());
			}
		}
		for (Enumeration e = inChannels.elements(); e.hasMoreElements() ; ){
			s.merge(((ChannelBuffer)e.nextElement()).getStatistics());
		}
		
	return s;
	}
	
	/**
	 * Returns the channel's source. By default, this is a dummy object.  This method is included for compatibility.
	 * NOTE: This will not return the ultimate source of the audio, since there
	 * is no single source.
	 * @return Source the channel's Source
	 */
	public Source getSource() {
		if (useOutputChannel)
			return outChannel.getSource();
		else 
			return source;
	}
	
	/**
	 * Sets the source of the Mixer channel to a single source. By default, MixerChannels have no real single
	 * source. Setting one could produce unpredictable results.
	 * @param s the new source for the MixerChannel
	 */
	public void setSource(Source s) {
		if  (useOutputChannel)
			 outChannel.setSource(s);
		else
			this.source = s;
	}

	/**
	 * Adds an additional source to the mixerChannel. 
	 * If a new input is added after the mixer has been opened, you must open the 
	 * input's corresponding channel manually by calling its open() method.
	 * @return Channel a channel to which you can assign a new source
	 */
	public Channel getNewInput(){
		
		ChannelBuffer in = new ChannelBuffer();
		inChannels.addElement(in);
		
		try {
			in.addDestination(new DummyDestination());
		} catch (DuplicateDestinationException dde) {}
	
		return in;
		
	}
	
	/**
	 * Adds an additional source to the mixerChannel that will be silent to the given destination.
	 * If the output is not a destination of the mixerChannel, it will be added as one.
	 * @param dest the Destination object to be removed from the mix sent to the returned channel
	 * @return Channel the new ChannelBuffer input to the mixer
	 */
	public Channel getNewSubtractedInput(Destination dest) {
		
		
		try {
			super.addDestination(dest);		//add dest as a destination for the mixer
		} catch(DuplicateDestinationException dde) {}
		Channel newChannel = getNewInput(); //get a new bufferChannel input to return
		ioLinks.put(dest,newChannel);		//record the linkage between input and output in iolinks vector
		return newChannel;
	}
	
	
	/**
	 * Closes and removes a channel form the mixer.
	 * <p>
	 * Closing this channel also closes its source.
	 * @param the channel corresponding to the input to be removed.
	 */
	public void removeInput(Channel c){	  
		//close the corresponding inChannel
		c.close();
		inChannels.removeElement(c);

	}
	/**
	 * Closes and removes from the mixer the channel corresponding to the given source.
	 * <p>
	 * (closes both channel and source)
	 * <p>
	 * <b>NOTE: slower than removeInput(Channel c).</b>
	 * @param s the Source corresponding to the input to be removed
	 */
	public void removeInput(Source s) {
		ChannelBuffer cb;
		for (Enumeration e = inChannels.elements();e.hasMoreElements();) {
			cb = ((ChannelBuffer)e.nextElement());
			if (cb.getSource() == s) {
				removeInput(cb);
			}
		}
	}
	
	/**
	 * Implements the mixer thread. Reads form all of the inputs, mixes the samples, and writes out the result
	 * to the destinations.
	 */
	public void run(){
		byte[] silence = new byte[mixSampleSize]; 
		for(int j = 0;j<silence.length;j++){silence[j] = (byte)128;} //init silence buffer
		
		Vector buffers = new Vector(); //vector to hold all of the samples to sum
		long startTime = System.currentTimeMillis();
		outBytes = 0;
		
		
		while (running) {
			
			try { 
				//for each input
				if (inChannels.isEmpty()) {
					try {
						//sleep 'till we need to send out another sample
						long sleeptime = (long)Math.max(((outBytes*SAMPLE_TIME)-(System.currentTimeMillis()-startTime))-BUFFER_TIME,0);
						readerThread.sleep(sleeptime);
					} catch (InterruptedException ie) {}
				}
				for (Enumeration e = inChannels.elements(); e.hasMoreElements() ; )
				{
					ChannelBuffer ch = ((ChannelBuffer)e.nextElement());
					byte[] data = new byte[mixSampleSize];
					
					
					
					//if data not available
					
					//Wait 'till we need to send
					long time = System.currentTimeMillis();
					try {
						//								 (desired elapsed     )-(actual        ) -time in buf
						long sleeptime = (long)Math.max(((outBytes*SAMPLE_TIME)-(time-startTime))-BUFFER_TIME,0);
												//System.out.println("***Sleeping for "+sleeptime);
						readerThread.sleep(sleeptime);
					} catch (InterruptedException ie){}
					if (((ChannelBuffer)ch).available() < mixSampleSize){
						totalSilence += mixSampleSize;
						if (ch.hasStarted()) ch.fellBehind(mixSampleSize);

					
					} else { 
						if (-1 == ((ChannelBuffer)ch).read(data, 0, mixSampleSize)) {
							//check for EOF
							throw new IOException();
						}
						buffers.addElement(data);  //store data into vector of samples
					}
					
				}
				int[] sum = null;
				byte[] sumBytes = null ;
				if (!buffers.isEmpty()){ 
					sum = this.sum(buffers); //sum the vector of samples
					if (running) { 
						//write mixed sample to all destinations
							
						if (useOutputChannel)
							outChannel.push(silence); //TODO: remove useOutputChannel stuff. No longer works
						else {
							for (Enumeration enu = destinations.elements();enu.hasMoreElements();) { 
								Destination dst = ((Destination)(enu.nextElement()));
								sumBytes = subtract(sum,dst,buffers);
								dst.write(sumBytes);
							}
						}
						
														//pw.println("Sending " + mixSampleSize + " byte sample at " + System.currentTimeMillis()+ ". Time allotted = "+waitTime);
						//debug...				
						//fos.write(sumBytes);
					}
				}
				
				else 
					if (running) { 
						//write silence to all destinations
							
						if (useOutputChannel)
							outChannel.push(silence);
						else {
							for (Enumeration enu = destinations.elements();enu.hasMoreElements();) { 
								((Destination)(enu.nextElement())).write(silence);
							}
						}
						
						
								//fos.write(silence);
	
				}
				outBytes+=mixSampleSize;
				buffers.removeAllElements(); //clear out the vector
			}
				
			catch (IOException e)
			{
				// Pipe has been broken, so quit...
				break;
			}
			
			catch (DataException de)
			{
				// FIXME: what to do with exceptions?
				de.printStackTrace();
			}
			
		}	
	}
	/**
	 * This method closes the MixerChannel. It first closes all of the channels inputs, 
	 * then brings down the mixer thread. finally, it closes the output buffer (if it exists)
	 */
	public void close(){
		if (closed) { 
			return;
		}
		closed = true;

		//close the inputs
		for (Enumeration e = inChannels.elements(); e.hasMoreElements() ; )
		{
			((ChannelBuffer)e.nextElement()).close();
		}
		//stop the mixer Thread
		
		if (running) {
			running = false;
			try {
				readerThread.join();
			} catch (InterruptedException ie) {}
			
		}
		//close the output stage
		if (useOutputChannel)
			outChannel.close();	
		
		/*
		try{pw.flush();fos.flush();fos.close();}catch(IOException e){}
			
		*/
	}
	
	/**
	 * This method will push data into the MixerChannel's output channel skipping over the mixer. 
	 * When sending data through the mixerChannel, data 
	 * should instead be sent by invoking the push methods of the inChannels. Included for compatibility.
	 * May produce unpredictable results.
	 * @param b the data to be sent to the output
	 */
	public void push(byte[] b) {
		if (useOutputChannel)
			outChannel.push(b);
		else 
			super.push(b);
	}
	/**
	 * Sum adds together the corresponding elements of the byte[]'s in the specified vector. 
	 * If the sum ever exceeds the max byte value, or falls below the min value, it is clipped.
	 * @param v a vector containing byte[]'s to be summed.
	 * @return byte[] the element by element summ of the vectors
	 */
	private int[] sum(Vector v) {
		int iSum;
		int[] bSum = new int[mixSampleSize];
		
		// init sum array to silence
		for(int j = 0;j<mixSampleSize;j++){bSum[j] = MicrophoneSource.SILENCE;}
		
		byte[] current;
		
		//for each channel
		for (Enumeration e = v.elements();e.hasMoreElements();) {
			byte[] inpt = ((byte[])e.nextElement());
			//for each sample
			for (int i=0;i<mixSampleSize;i++){	

				//add
				//if ( inpt[i] != -1) {
					bSum[i] = (bSum[i] & 255)+ (inpt[i]& 255) - (MicrophoneSource.SILENCE&255); 
					
					//clip to rails
					/*
					if (iSum < 0 ) bSum[i] = 0;
					else if (iSum > 254) bSum[i] = (byte)254;
					else bSum[i] = (byte)iSum;
					*/
				//}
			}

							
		}
	
		return bSum;

	}
	
	private byte[] subtract(int[] sum,Destination dest,Vector inputs) {
		//TODO: Syncronization issue? We may drop the wrong voice for one sample when a participant drops.
		byte[] result = new byte[sum.length];
		int iSum;
		Channel inch = (Channel)ioLinks.get(dest);
		if (inch != null)
		{
			int index = inChannels.indexOf(inch);
			if (index !=-1) {
				byte[] toSubtract;
				try {	
					toSubtract = (byte[])inputs.elementAt(index);
				}catch (ArrayIndexOutOfBoundsException aiobe) {
					toSubtract = null;
				}
				if (toSubtract != null) {
					for(int i=0;i<sum.length;i++) {
						iSum = sum[i] - (toSubtract[i]& 255) + (MicrophoneSource.SILENCE & 255);
						//clip to rails:
						if (iSum < 0 ) result[i] = 0;
						else if (iSum > 254) result[i] = (byte)254;
						else result[i] = (byte)iSum;
						
						
					}
					return result;
				}
			}
		}
		
		//we don't need to subtract anything. just convert to bytes.
		for(int i=0;i<sum.length;i++) {
			//clip to rails:
			if (sum[i] < 0 ) result[i] = 0;
			else if (sum[i] > 254) result[i] = (byte)254;
			else result[i] = (byte)sum[i];
							
							
		}
		return result;
		
	}

}

/**
 * Looks like a channel, actually just a buffer. These are returned on a call to MixerChannel.getNewInput().
 * They contain all of the channel functionality that a source expects, and buffer data to be read out by the mixer.
 */
class ChannelBuffer extends Channel{
	
	/**
	 * Has our source started sending yet?
	 */
	private boolean started;
	/**
	 * The time when the first push was executed
	 */
	private long firstPushTime;
	/**
	 * The total number of bytes read out of the buffer
	 */
	private int totalRead;
	/**
	 * The number of bytes our reader dropped from us that we need to eliminate
	 */
	private int behind;
	/**
	 * The input stream used to buffer the data.
	 */
	private PipedInputStream pis;
	/**
	 * The output stream used for buffering data.
	 */
	private PipedOutputStream pos;

	/**
	 * The size of the internal buffer. Currently, BigPipedInputStream has a buffer of 6400 bytes. 
	 * We can use up to this amount.
	 */
	private int BUFFER_SIZE = 6400;  //the size of the buffer in the PipedInputStream
	/**
	 * Samples/second
	 */
	private int SAMPLE_RATE = MicrophoneSource.SAMPLE_RATE;
	/**
	 * How many consecutive bytes are we willing to cull from a sample to try to catch up
	 */
	private int MAX_CATCHUP = 10;
	
	/**
	 * True if the last attempt to read from this channel met an empty buffer
	 */
	private boolean isDelinquent = false;
	/**
	 * Number of instances of the channelbuffer; used for statistics output numbering.
	 */
	static int instances = 0;
	
	/**
	 * Instance number of the current channelbuffer; used for stat output numbering.
	 */
	private int iMyInstance;

						private PrintWriter pw;
							
	/**
	 * Creates a new ChannelBuffer. Allocates the internal buffer.
	 */
	public ChannelBuffer(){
		running = false;
		closed = false;
		
		this.iMyInstance = instances++;
		try{

			pis = new BigPipedInputStream();
			pos = new PipedOutputStream(pis);
						//FileOutputStream fos = new FileOutputStream("cb"+iMyInstance+".txt");
						//if (fos == null) System.exit(32453);
						//pw = new PrintWriter(fos);
						//pw.println("--Start--");
		}
		catch (IOException e)
		{
			e.printStackTrace();
		}
	}


	/**
	 * Starts the channel's source
	 */
	public void open() {
		source.start();
	}
	
	/**
	 * Method for the source to feed data to the BufferChannel. When the buffer runs out of space for a new push()
	 * we throw away just enough old data to make room for the new.
	 * @param b the byte array to be buffered
	 */
	public void push(byte[] b){
		try{
						long time = System.currentTimeMillis();
						//pw.println(" "+iMyInstance+" beginning Push at "+time); 
			if (!started) {
				started = true;
				firstPushTime = System.currentTimeMillis();
			}
			
			synchronized (this) {
				int catchBy = Math.max(Math.min(MAX_CATCHUP,behind),0);
				//if the buffer is too full for this sample
				if (available()+b.length - catchBy > BUFFER_SIZE) {
					//ditch just enough old data so this new sample will fit
					int catchup = Math.max(b.length-(BUFFER_SIZE-available()),0);
					behind -= catchup;
							//pw.println(" Culling out " + catchup+" bytes of old data for push "+ available()+" at "+System.currentTimeMillis());			
					pis.read(new byte[catchup]);
					
				}
				if (available() > /*behind >*/2048) {
					//trim samples out of packet to catch up
					pos.write(b,0,b.length-catchBy);
					
					/*
					pos.write(b,0,b.length/4 - catchBy);
					pos.write(b,b.length/4,b.length/4 - catchBy);
					pos.write(b,2*b.length/4,b.length/4 - catchBy);
					pos.write(b,3*b.length/4,b.length/4 - catchBy);
					*/
					
					behind -= catchBy;
									//pw.println(" caught up by "+catchBy+ ". now behind by "+behind);
				} else {
			
			
			
			
					pos.write(b);
			
							//long elapsed = System.currentTimeMillis() - time;
							//		if (elapsed > 0) 
							//pw.println(" "+iMyInstance+" **** Blocked push took "+elapsed+"ms Available: "+this.available());
							//		else
							//pw.println(" "+iMyInstance+" Push took "+elapsed+"ms Available: "+this.available());
				}
			}
			//sanity check
			//if (behind <0) throw new NullPointerException("behind by negative number");
			
		} catch (IOException e) {}
	}
	
	/**
	 * Method for the mixer to read data out of the buffer. Reads out only one byte
	 */
	public int read() throws IOException{
		if (pis.available() < 1) { 
			long time = System.currentTimeMillis();
							//pw.println(iMyInstance+"beginning read at "+time);
			int temp = pis.read();
							//pw.println(iMyInstance+"read blocked for "+(System.currentTimeMillis()-time)+"ms" );
			return temp;
		}
							//pw.println(iMyInstance+"Unblocked read. Available: (pre) "+ this.available()+" at "+System.currentTimeMillis());
		totalRead++;
		return pis.read();
	}
	
	/**
	 * Method for the mixer to read data out of the buffer. Trys to fill the given byte array 
	 * with data from the buffer.
	 * @param b the buffer to grab data into
	 */
	public int read(byte[] b) throws IOException{
		totalRead+= b.length;
		return pis.read(b);
	}
	
	/**
	 * Method for the mixer to read data out of the buffer.
	 * @param b The array where the data will go
	 * @param p1 the starting index in b
	 * @param p2 the number of bytes to read
	 * @see java.io.PipedInputStream.read
	 */
	public int read(byte[] b, int p1, int p2)throws IOException{
		 
			long time = System.currentTimeMillis();
							//pw.println(iMyInstance+"beginning blocked read at"+time);
			int temp = pis.read(b,p1,p2);
							//pw.println(iMyInstance+"read of "+(p2-p1)+" blocked for "+(System.currentTimeMillis()-time)+"ms. avail="+this.available());
			totalRead+=(temp);
			return temp;
							
	}
	
	/**
	 * Returns the number of bytes left in the buffer.
	 * @exception IOException thrown when the PipedOutputStream that forms the buffer is broken.
	 */
	public int available() throws IOException {
		return pis.available();
	}
	
	
	/**
	 * Closes the BufferChannel. Closes the channel's source and buffer.
	 */	
	public void close() {
		if (closed) {
			return;
		}
		try {
			pos.close();
			pis.close();
		} catch (IOException e) {}
		if (source != null) {
			source.close();
		}
		closed = true;
	}
	/**
	 * Gets the contents of isDelinquent.
	 * @return boolean the current state of isDelinquent. 
	 */
	public boolean getDelinquent() {
		return isDelinquent;
	}
	
	
	/**
	 * Sets the contents of isDelinquent. This is called with true whenever we need to read,
	 * but the buffer does not have enough data.
	 * @param delinq the value setDelinquent will set isDelinquent to.
	 */
	public void setDelinquent(boolean delinq) {
		isDelinquent = delinq;
	}
	
	/**
	 * Returns true if the source has sent data to this channel.
	 */
	public boolean hasStarted() {
		return started;
	}
	
	/**
	 * Returns the number of ms early we are. If this is negative, we are late. 
	 * This method assumes that we are late if our samples read by read() method/time elapsed is < SAMPLE_RATE.
	 */
	public long getEarlyTime() {
		if (started) { 
			//desired elapsed time - actual elapsed time
			long early = totalRead/(SAMPLE_RATE/1000)-(System.currentTimeMillis()-firstPushTime);
							//pw.println("Early by " +early);
			return early;
		}else return (long)0;
		
	}
	/**
	 * lets the reading channel inform the bufferChannel that it dropped bytes when our buffer was empty.
	 */
	public void fellBehind(int by) {
		synchronized (this) {
			behind += by;
		}
							//pw.println("fell behind an additional "+by+" to "+behind+" at "+System.currentTimeMillis());
	}
	
	/**
	 * Returns the statistics for this ChannelBuffer, and its destination.
	 * @return Stats the stats for this ChannelBuffer -- currently, the number of bytes in the buffer.
	 */
	public Stats getStatistics(){ 
	
		Stats s = new Stats();
		try {
			s.addStat("Device","ChannerBuffer");
			s.addStat("<Channel Buffer "+iMyInstance+"> Bytes in Buffer ",new Integer(this.available()));
			s.addStat("<Channel Buffer "+iMyInstance+"> Total bytes read from me",new Integer(totalRead));
			s.addStat("<Channel Buffer "+iMyInstance+"> behind by",new Long(behind) );
		} catch (IOException e) {e.printStackTrace();}
		s.merge(super.getStatistics());
		return s;
	}
	
		//HACK: we should make our own 6400 byte buffer...
	private class BigPipedInputStream extends PipedInputStream {
		protected int PIPE_SIZE = 6400;
		public BigPipedInputStream() {
			super();
			PIPE_SIZE = 6400;
			buffer = new byte[PIPE_SIZE]; 
		}
	}
}

