/* ***************************************************************************+
 * 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.devices;

import java.io.*;
import cnrg.itx.datax.*;

/**
 * Class implementing a stream source. This class will take any InputStream and
 * synchronize it to make it a valid <code>Source</code>.
 */
public class StreamSource implements Source, Runnable
{
	/**
	 * Thread start/stop flags
	 */
	private boolean start;
	
	/**
	 * Attribute to store the input stream represented by this source.
	 */
	private InputStream inputStream;
	
	/**
	 * Attribute for storing the stream statistics.
	 */
	private Stats streamStats;
	
	/**
	 * Attribute for storing the Channel reference.
	 */
	private Channel channel;
	
	/**
	 * Attribute for storing the device instance number. This is a static attribute and will be
	 * incremented in the constructor. This represents the total number of instances of
	 * the stream source.
	 */
	private static int instances = 0;
	
	/**
	 * Attribute for storing the device ID. This is the value of the instance at the moment.
	 */
	private int iDevice;
	
	/**
	 * Attribute for storing the number of bytes written by this source.
	 */
	private int bytesWritten = 0;
	
	/**
	 * Attribute to store the thread object.
	 */
	private Thread thread;
	
	/**
	 * How long should the transfer thread wait between writes?
	 */
	private long waitTime;
	
	/**
	 * How many buffers should the transfer thread keep?
	 */
	private int bufferSize;
	
	/**
	 * How many samples are currently buffered?
	 */
	private int numBuffered;
	
	/**
	 * Record of last call to System.currentTimeMillis()
	 */
	private long startTime;
	
	/**
	 * Mute state
	 */
	private boolean mute;
	
	/**
	 * File sample size
	 */
	public static final int SAMPLE_SIZE  = 400;
	
	/**
	 * Default buffer time
	 */
	public static final int DEFAULT_BUFFER_TIME = 100;
	
	/**
	 * Constructor to make a StreamSource from any input stream (no flow control)
	 * @param inputStream The input Stream
	 * @param channel The audio channel
	 */
	public StreamSource(InputStream inputStream, Channel channel)
	{
		this(inputStream, channel, DEFAULT_BUFFER_TIME);
	}
	
	/**
	 * Constructor to make a StreamSource from any input stream (flow control)
	 * @param inputStream The input Stream
	 * @param channel The audio channel
	 * @param bufferTime The amount of data to keep buffered on the output stream in ms
	 */
	public StreamSource(InputStream inputStream, Channel channel, long bufferTime)
	{
		// Increment the number of instances and assign the device id
		instances++;
		iDevice = instances;

		this.inputStream = inputStream;
		this.channel = channel;
		
		// Create a new strem stats object
		streamStats = new Stats();
		
		// Initialize thread flag
		start = false;
		
		// Flow control calculations:
		//((samples/buffer) * (ms/sec)) / (samples/sec) = ms/buffer
		waitTime = (SAMPLE_SIZE * 1000) / MicrophoneSource.SAMPLE_RATE;
		// (samples/ms) * ms = samples
		bufferSize = (int)(((MicrophoneSource.SAMPLE_RATE/1000) * bufferTime) / SAMPLE_SIZE);
		numBuffered = 0;
	}	
	
	/**
	 * Method to close the input stream.
	 */
	public void close()
	{
		// Stop the thread
		stop();
		
		// Close the input stream
		try
		{
			inputStream.close();
		}
		catch(IOException e)
		{
			// Ignore for now..
		}
	}
	
	/**
	 * Method to mute the input stream. This method will either block or resume
	 * the stream according to the value passed to it.
	 * @param mute true to mute and false to resume the stream
	 * @return boolean The previous mute state
	 */
	public boolean mute(boolean mute)
	{
		this.mute = mute;
		return mute;
	}
	
	/**
	 * Method to get the statistics of the input stream.
	 * @return Stats The statistics object
	 */
	public Stats getStatistics ()
	{
		return streamStats; 
	}
	
	/**
	 * Method to start the source thread.
	 */
	public void start()
	{
		if (start)
			return;
		
		start = true;
		
		thread = new Thread(this);
		thread.start();
	}
	
	/**
	 * Method which acts as a thread and pushes data to the channel.
	 */
	public void run()
	{
		long interval = 0;
		long newWaitTime;
		int retval = 0;
		byte[] audioBuffer = new byte[SAMPLE_SIZE];
		
		// Initialize the system time for flow control purposes
		startTime = System.currentTimeMillis();
		
		// Are we done?
		try
		{
			while (start)
			{
				// Try to read in some bytes. Note that not all the bytes may be read
				// at one go.
				retval = inputStream.read(audioBuffer, 0, SAMPLE_SIZE);
				
				// Break out if there is no more data
				if (retval == -1)
				{
					break;
				}
				
				// Flow control algorithm
				numBuffered++;
				newWaitTime = Math.max((waitTime - (System.currentTimeMillis() - startTime)), 0);
				if (numBuffered > bufferSize)
				{
					if (newWaitTime < 0)
					{
						// Will not count time from last buffer!  --> Therefore catches up.
					}
					else
					{
						// Wait until it is possible to send again.
						try
						{
							this.thread.sleep(newWaitTime);
						}
						catch (InterruptedException e)
						{
							return;
						}
					}
				}
				else
				{
					if (waitTime > 0)
					{
						// Wait for half the time to begin filling buffer on remote end.
						try
						{
							this.thread.sleep(newWaitTime / 2);
						}
						catch (InterruptedException e)
						{
							System.out.println("<StreamSource> --> thread exiting");
							return;
						}
					}
				}
					
				// Get the new system time
				startTime = System.currentTimeMillis();
				
				// Push the data into the channel if the device is not muted
				if (!mute)
				{
					channel.push(audioBuffer);
					
					bytesWritten += audioBuffer.length;
					streamStats.addStat("<Stream Source " + iDevice + "> Bytes written ", new Integer(bytesWritten));
				}
			}
		}
		catch(IOException e)
		{
		}
	}

	/**
	 * Method to stop the thread.
	 */
	public void stop()
	{	
		if (start == false)
			return;
		
		// Stop the active thread
		start = false;
		
		// Wait for the thread to join
		if (thread != null && thread.isAlive())
		{
			try
			{
				thread.join();
			}
			catch(InterruptedException e)
			{
				e.printStackTrace();
			}
		}
		
		thread = null;
	}
	
	/**
	 * Returns a collection of properties supported.
	 */ 
	public PropertiesCollection getProperties() throws DataException
	{
		return null;
	}

	/**
	 * Sets the given properties collection into the device
	 */
	public void setProperties(PropertiesCollection pc) throws DataException
	{
	}

	/**
	 * 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
	{
	}
}
