/* ***************************************************************************+
 * 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.net.*;
import java.io.*;
import java.util.*;
import cnrg.itx.datax.*;

/**
 * Class to implement the network source. The network source acts as the network endpoint
 * of a connection. It listens to the network pushes data into the channel.
 */
public class NetworkSource implements Source, Runnable
{
	/**
	 * Attribute to store the channel for this source.
	 */
	private Channel channel;
	
	/**
	 * Attribute for the netowrk socket which acts as the actual source of data.
	 */
	DatagramSocket socket;
	
	/**
	 * Thread start/stop flags
	 */
	private boolean start;
	private boolean mute = false;
	
	/**
	 * 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;
	
	/**
	 * Sample size.
	 */
	private int iSampleSize;
	
	/**
	 * Attribute for storing the number of bytes written by this source.
	 */
	private int bytesWritten = 0;

	/**
	 * Attribute storing number of packets received for this source
	 */
	private int iNumPackets = 0;
	
	/**
	 * Attribute to store the thread object.
	 */
	private Thread thread;

	/**
	 * Buffer to reorder and provide flow control for RTP packets. The bufferQueue is
	 * created when we know the other person's inital sequence number.
	 */
	BufferQueue bq;
	
	/**
	 * Pulls data from the buffer queue and feeds into the channel.
	 */
	ChannelFeeder cf;
	
	/**
	 * Allows byte arrays to be converted to rtpPackets
	 */
	RTPPacketFactory rtpFactory;
	
	/**
	 * Holds a the silence packet
	 */
	byte [] bSilence;
	
	/**
	 * Attribute for sprcifying if the FEC and RTP should be used.
	 */
	private boolean useFECandRTP = true;
	
	/**
	 * Constructs a Network object. Lets java find the port.
	 * <p>
	 * This is the preferred constructor for this object.
	 * <p>
	 * NOTE: Sample size must be EQUAL across remote destination and local source!
	 * @param channel the Channel that this newtwork source is pushing into
	 * @param iSampleSize the sample size of data sent over the network
	 * @exception DataException thrown if no port is free on this computer
	 */
	public NetworkSource(Channel channel, int iSampleSize) throws DataException
	{
		// Store the channel object
		this.channel = channel;
		
		// Create a new statistics object to store the netowrk statistics

		// Increment the number of instances and assign the device id
		instances++;
		iDevice = instances;

		// Create a new datagram socket and make it bind to some local port
		try
		{
			socket = new DatagramSocket();
		}
		catch (IOException ioe)
		{
			throw new DataException ("Could not acquire a free port");
		}

		// Set the sample size
		this.iSampleSize = iSampleSize;
		
		// create the new rtpFactory
		rtpFactory = new RTPPacketFactory();
		
		// create silence packet
		bSilence = new byte[ iSampleSize ];
		for (int i = 0; i < iSampleSize; i++)
		{
			bSilence[i] = MicrophoneSource.SILENCE;
		}
		
		// Initialize thread flags
		start = false;
	}
	
	/**
	 * Constructs a Network object. Lets java find the port.
	 * <p>
	 * This is the preferred constructor for this object.
	 * <p>
	 * NOTE: Sample size must be EQUAL across remote destination and local source!
	 * @param channel the Channel that this newtwork source is pushing into
	 * @param iSampleSize the sample size of data sent over the network
	 * @param useFECandRTP true if FEC and RTP have to be used
	 */
	public NetworkSource(Channel channel, int iSampleSize, boolean useFECandRTP) throws DataException
	{
		this (channel, iSampleSize);
		this.useFECandRTP = useFECandRTP;
	}

	/**
	 * Method to start the network source.
	 */
	public void start()
	{
		if (start)
		{
			return;
		}

		// Initialize the thread flags to reflect the active thread
		start = true;
		
		// Create the new thread and activate it
		thread = new Thread(this);
		thread.start();
	}

	/**
	 * Method which acts as a thread and pushes data to the channel.
	 */
	public void run()
	{
		//fixme; remove constants and change to iSampleSIze to remove fec
		// comment the next line, and uncomment the next to remove fec buffer
		byte[] networkBuffer = null;
		
		if (useFECandRTP)
			networkBuffer = new byte[iSampleSize * 3 + 48];
		else
			networkBuffer = new byte[iSampleSize];
		
		DatagramPacket packet = new DatagramPacket(networkBuffer, networkBuffer.length);
		
		try
		{
			if (useFECandRTP && bq != null)
				cf = new ChannelFeeder(); 
			
			while(start)
			{
				// Wait for a packet to come on the receiving port
				socket.receive(packet);
				iNumPackets ++;
				
				// We might need to wait for some time before pushing
				// the data to the channel.
				
				// Push the data into the channel if the device is not muted
				if(!mute)
				{
					if (useFECandRTP)
						bq.put( rtpFactory.getRTPPacket(networkBuffer));
					else
						channel.push(networkBuffer);
						bytesWritten+= iSampleSize;
				}
			}
			
			if (useFECandRTP)
				cf.bRunning = false;
		}
		catch(IOException e)
		{
			// The connection might have been closed from the other side.
		}
	}
	
	/**
	 * Method to stop a netowrk source.
	 */
	public void stop()
	{
		if (!start)
		{
			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();
			}
		}
	}
	
	/**
	 * Method to close a network source.
	 */
	public void close()
	{		
		// Close the network socket
		socket.close();
		
		// Stop the thread
		stop();	
	}

	/**
	 * Method to mute the network source.
	 * @param mute true if the device has to be muted and false if it has to be activated
	 * @return boolean true if the device was muted and false if it was activated
	 */
	public boolean mute(boolean mute)
	{
		this.mute = mute;
		return mute;
	}
	
	/**
	 * Method to get the statistics from the netowrk source
	 * @return Stats the statistics object
	 */
	public Stats getStatistics ()
	{
		Stats networkStats = new Stats();
		networkStats.addStat("Device", "Network source");
		networkStats.addStat("<Network Source " + iDevice + "> Bytes written to channel", new Integer(bytesWritten));
		networkStats.addStat("<Network Source " + iDevice + "> Number Packets Received from wire", new Integer(iNumPackets)); 

		if (useFECandRTP && bq != null)
			networkStats.merge(bq.getStatistics ());
		
		return networkStats;
	}
	
	/**
	 * Returns a collection of properties supported.
	 * @return a collection of device properties
	 * @exception DataException thrown if the device properties cannot be defined
	 */ 
	public PropertiesCollection getProperties() throws DataException
	{
		PropertiesCollection pc = new PropertiesCollection();
		try 
		{
			NetworkProperty p = null;
			if (useFECandRTP)
				p = new NetworkProperty(socket.getLocalPort(),
										InetAddress.getLocalHost(), 
										iSampleSize, 0, 0);
			else
				p = new NetworkProperty(socket.getLocalPort(),
										InetAddress.getLocalHost(), 
										iSampleSize, 0, -1);
			pc.addProperty(p);
		}
		catch (UnknownHostException uhe)
		{
			uhe.printStackTrace();  
		}
		return pc;
	}

	/**
	 * Sets the given properties collection into the device
	 * @param pc the set of properties to be set on the device
	 * @exception DataException thrown if the provided proerties cannot be set on the device
	 */
	public void setProperties(PropertiesCollection pc) throws DataException
	{
		// nothing needs to be done here.
	}

	/**
	 * 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
	{
		boolean bGotProperty = false;
		Object prop;
		
		for (Enumeration e = pc.getProperties(); e.hasMoreElements() ; )
		{
			prop = e.nextElement();
			
			if (prop instanceof NetworkProperty)
			{
				if (! bGotProperty)
				{
					if (((NetworkProperty)prop).getInitialSequenceNumber () != -1)
						bq = new BufferQueue( iSampleSize / 8 , iSampleSize, ((NetworkProperty)prop).getInitialSequenceNumber());
					else
						useFECandRTP = false;
					
					bGotProperty = true;
				}
				else 
				{
					throw new DataException("More than 1 network destination sending data to us!");
				}
			}
		}
	}

	/**
	 * Private class to implement a channel feeder.
	 */
	class ChannelFeeder extends Thread
	{
		boolean bRunning;
		
		ChannelFeeder()
		{
			bRunning = true;
			this.start();
		}
		
		public void run()
		{
			byte [] bData;
			while ( bRunning )
			{
				bData = NetworkSource.this.bq.get();
				
				// Increment the number of bytes in the channel and reflect it
				// in the network statistics.
				NetworkSource.this.bytesWritten += (bData == null ? 0 :  bData.length);
				if (bData != null)
					NetworkSource.this.channel.push(  bData == null ? NetworkSource.this.bSilence : bData );
			}
		}
	}
}