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


/**
 * This class will implement a buffered queue for reordering incoming
 * RTP Packetes.  The class will allow packets to be removed in a
 * blocking read which waits until "packet boundries" until serving
 * out the data.  
 * <p> 
 * RTP packets are put into the queue via the
 * <code>put()<\code> method. It will check to see whether we already
 * have the sample's data (via a FEC packet).  If so, we don't need to
 * do anything. Otherwise, we should depack the packet and add it into
 * the queue in the proper order. We should fill in any preceeding
 * sample "holes" -- places where we should have received a packet but
 * didn't -- with silence samples. If we later find this packet, the
 * silence sample should be replaced with the real data.  Finally, the
 * put method should check whether the queue is growing too large
 * (which indicates that data isn't being read) and "clean" out old
 * samples by deleting them.
 * <p> Data samples are removed from the
 * queue via the get method. The get method will block until a packet
 * boundary, and then return a sample.  The sample will either be the
 * "correct" sample for that time period, or a silence sample if the
 * correct sample is not available (and FEC could not fill in the
 * missing information. For smooth playback, <code>get()</code> should be
 * called in succession with minimal processing.
 */ 
public class BufferQueue implements Statistics
{
	/**
	 * The sequence number of the last sample that was read */
	private int iLastPacket ;

	/**
	 * The difference between our time and the other time in milliseconds. 
	 * <p> DELTA is calculated by taking the other guy's time - our time
	 */
	private int iDeltaTime ;

	/**
	* The current buffer size to use.  The buffer size determines, in
	* part, the amount of latency that the connection uses. It is in
	* terms of the number of samples to buffer.
	*/
	//	private int iBufferSize;

	/**
	 * Amount of latency to build into the buffer.
	 */
	private int iBufferLatency;

	/**
	 * The amount of TIME in millisecodns that each sample represents.
	 */
	private int iSampleTime;

	/**
	 * The current size of the queue (number of samples)
	 */
	private int iSize;

	/**
	* Semaphore to show ownership of who gets to change the queue.
	*/
	//	private Object semaphore;

	/**
	 * Head of the queue
	 */
	private SampleNode snHead;

	/**
	 * Tail of the queue
	 */
	private SampleNode snTail;
	
	/**
	 * Attribute for storing a FECBuffer object
	 */
	private FECBuffer fecBuffer;

	/**
	 * The starting buffer size.
	 */
	final static int INITIAL_BUFFER_LATENCY = 100;

	/**
	 * one half of the sequence number space.
	 */
	final static int SEQUENCE_SPACE = (2 << 15);
	final static int HALF_SEQUENCE_SPACE = (SEQUENCE_SPACE >> 1);

	/**
	 * Number of instances of the buffer queue; used for statistics output numbering.
	 */
	static int instances = 0;
	
	/**
	 * Instance number of the current buffer queue; used for stat output numbering.
	 */
	private int iMyInstance;

	/**
	 * Number of null packets delivered.
	 */
	private int iNullSamples = 0;

	/**
	 * Number of samples removed (because too old).
	 */
	private int iRemovedSamples = 0;

	/**
	 * Number of samples removed (because too old).
	 */
	private int iNumOldPacketsReceived = 0;

	/**
	 * Number of samples inserted that had to be reordered
	 */
	private int iReordered = 0;

	/**
	 * Constructor for the BufferQueue.
	 * @param iSampleTime the reference sample time for the buffer queue
	 */
	public BufferQueue(int iSampleTime, int iSampleSize, int iLastPacket)
	{
		iMyInstance = instances++;
		
		this.iLastPacket = iLastPacket - 1 ;
		// shift so that sequence space is always positive
		this.iLastPacket += this.iLastPacket < 0 ? SEQUENCE_SPACE: 0;
		this.
			 iDeltaTime = Integer.MAX_VALUE;
		iBufferLatency = INITIAL_BUFFER_LATENCY;
		iSize = 0;
		this.iSampleTime = iSampleTime;
		
		// Make a FEC buffer to manage the buffers
		fecBuffer = new FECBuffer(iSampleSize);

		snHead = new SampleNode(null, -1, 0);
		snHead.bPriorDataSequenceComplete = true;
		snTail = new SampleNode(null, -2, 0, true);
		
		snHead.later = snTail;
		snTail.earlier = snHead;      

		//		System.out.println("BUFFER_QUEUE Initialization: SampleTime: " + iSampleTime);
		//		System.out.println("                             SampleSize: " + iSampleSize);
		//		System.out.println("                             LastPacket: " + this.iLastPacket);
	}
	
	/**
	 * Returns whether sequence iB is older than or equal to iA. This method takes
	 * care of the fact that sequence numbers can wrap around and then a
	 * smaller sequence number can be newer than a larger sequence number.
	 * @param iA the first sequence number
	 * @param iB the second sequence number
	 */
	private boolean isOlder(int iA, int iB)
	{
		// Get the difference of the two sequence numbers
		int iBMinusA = Math.abs(iA - iB);
		
		// If the first sequence number is less than the second then it
		// is obvious that the second one is the new one. If the second
		// sequence number is lesser than the first then it is not very
		// clear if it is an old one or if the sequence number has wrapped
		// around. Just like in TCP sequence numbers, we assume that if
		// the new sequence number is less than half the sequence number
		// space, then it is a wrap around and the smaller number is the
		// new one.
		return (( iA <= iB && iBMinusA < HALF_SEQUENCE_SPACE) ||
				( iA >= iB && iBMinusA > HALF_SEQUENCE_SPACE));
	}

	/**
	 * Checks if a particular sample should be put into the queue.
	 * @param iSequenceNumber the sequence number of the packet to check
	 * @return SampleNode returns a null if the node shouldn't be added or
	 *                    else it returns the location where to insert it
	 */
	private SampleNode needToAdd(int iSequenceNumber)
	{
		//FIXME: delete: debug only
		if (isOlder(iSequenceNumber, iLastPacket - 2))
			iNumOldPacketsReceived ++;

		// Sample is way too old so it shouldn't be put in the queue
		if (isOlder(iSequenceNumber, iLastPacket))
		{
			return null;

		}

		// Sample is new. We might need to add some silence samples between
		// last in queue and the new guy
		if (isOlder(snTail.earlier.iSequenceNumber, iSequenceNumber) && snTail.earlier.iSequenceNumber != iSequenceNumber)
			return snTail.earlier;
		
		// Else scan through to look for location 
		SampleNode sn = snTail.earlier;
		
		// Look through the whole queue starting from the tail till we
		// reach the head or till we hit a node till which the sequence
		// is complete
		while (sn != snHead && ! sn.bPriorDataSequenceComplete)
		{
			// Found it
			if (sn.iSequenceNumber == iSequenceNumber)
			{
				// if we filled in data with silence, replace
				if (sn.bIsSilence)
					return sn;
				else
					return null;     // We have valid data. Do nothing
			}
			
			// Move backwards to find it
			sn = sn.earlier;      
		}
		
		// Should be added to front.
		if (sn == snHead)
		{
			return snHead;
		}
		
		// No need to scan any further. We know that it is there.
		return null;
	}

	/**
	 * Adds a buffer to the queue.
	 * @param sample byte array of the actual data
	 * @param iRTPSequence sequence number of the sample
	 * @param lTimeStamp time stamp that the sample was generated.
	 */
	private void addToQueue(byte[] sample, int iRTPSequence, long lTimeStamp)
	{
		// Find out if the node should be added
		SampleNode snLocation = needToAdd(iRTPSequence);
		
		// Shouldn't be added so return
		if (snLocation == null)
		{
			return;
		}
		
		// There will be two cases....
		
		// Case 1: The packet had not come in before and silence was filled
		// in its place. This needs to be replaced with the new data.
		if (snLocation.iSequenceNumber == iRTPSequence)
		{
			iReordered ++;
			// Replace the silence packet with the data
			snLocation.bData = sample;
			
			// Silence flag is set to false as there is valid data in it
			snLocation.bIsSilence = false;
			
			// Set the new location for the completed packet sequence
			// in the queue. This is required as there might be one packet
			// missing the queue and that would complete the whole sequence
			// till another node. So we need to go forward and mark all
			// the nodes as completed till we reach the tail or a silence
			// packet. We should also do this only if the previous node
			// is not a slience node because then the sequence will not
			// be completed till the previous packet.
			if (snLocation.earlier.bPriorDataSequenceComplete)
			{
				// Traverse till the end of the queue
				while (! snLocation.bIsSilence)
				{
					// Set the completed flag to true
					snLocation.bPriorDataSequenceComplete = true;
					
					// Move ahead
					snLocation = snLocation.later;
				}
			}
		} 
		else // 2: THis is a new packet; we may need to add a whole bunche of silence stuff.
		{
			// If it's empty, need to call a notify in case the get is waiting on us to put in data
			boolean bIsEmpty = iSize == 0; 

			// Find number of packets to add in (takes care of wraparound). The previous 
			// guy's sequence number; make an exception for the head 
			int iSequenceStart = (snLocation == snHead ?  iLastPacket : snLocation.iSequenceNumber);
			int iNumPacket = iRTPSequence - iSequenceStart ;
			if (iNumPacket < 0)
			{
				iNumPacket += SEQUENCE_SPACE ;
			}
			SampleNode snSilence;

			// puts in the silence
			for (int i = iSequenceStart +1; i < iSequenceStart +iNumPacket  ; i++)
			{
			   //				System.out.println ("Putting in silence at " + i % SEQUENCE_SPACE);
				// silence == null!!
				snSilence = new SampleNode(null, i % SEQUENCE_SPACE,  
										   lTimeStamp - iSampleTime * (i - snLocation.iSequenceNumber), 
										   true);
				snSilence.earlier = snTail.earlier;
				snSilence.later = snTail;
				snTail.earlier = snSilence;
				snSilence.earlier.later = snSilence;
			}
			
			iSize += iNumPacket ;

			// add in the actual data packet
			SampleNode snNew = new SampleNode(sample, iRTPSequence, lTimeStamp);
			snNew.earlier = snTail.earlier;
			snNew.later = snTail;
			snTail.earlier = snNew;
			snNew.earlier.later = snNew;
			
			// if anyone is waiting on the empty queue, notify them.
			if (bIsEmpty)
			{
				snHead.notify();
			}	
		}
	}

	/**
	 * Puts a buffer into the queue. Forward error correction is performed
	 * on the buffer and then it is put in the queue.
	 * @param rtp the RTPPacket to be inserted into the queue
	 */
	public void put (RTPPacket rtp)
	{
		// This operation changes the buffer data sructure which might also
		// be operated on by the get method.
		synchronized (snHead)
		{
			// Get the RTP header from the packet
			RTPHeader rtph = rtp.getRTPHeader();

			// Get the sequence number from the header. Convert it to an
			// integer so that the calculations later can be done easily
			int iRTPSequence = rtph.getSequence();

			// Get the time stamp of the RTP packet
			long lTimeStamp = rtph.getTimeStamp ();

			// Adjust the time delta
			int iNewDelta = (int) ((int) System.currentTimeMillis () - lTimeStamp);

			if (iNewDelta <  iDeltaTime)
			{
				iDeltaTime = iNewDelta;
				//				System.out.println ("Delta now: " + iDeltaTime);
			}

			// Unpack the RTP packet into all the buffers it has
			fecBuffer.unpack (rtp.getData ());
			
			// Get all the buffers from and add them to the queue

			for (int i=fecBuffer.getNumFECPackets ()-1; i>=0 ; i--)
			{
				addToQueue (fecBuffer.getBuffer (i), (iRTPSequence-i+SEQUENCE_SPACE) % SEQUENCE_SPACE, 
							lTimeStamp-i*iSampleTime);
			}

			// FIXME: check for too large queue size (How to do??)
		}
	}

	/**
	 * Gets the next sample from the queue. Blocks until a time boundary
	 * unless we are "behind". returns null for silence;
	 */
	public byte[] get()
	{
		byte [] bData = null;
		synchronized(snHead)
		{
			if (iSize  == 0) 
			{
				// FIXME: we should have some sort of time out or something...
				try
				{	
					snHead.wait(iSampleTime);
				}
				catch (InterruptedException ie)
				{
					ie.printStackTrace();
				}
			}
			if (iSize == 0)
			{
				iNullSamples ++;
				//				System.out.println("QUEUE SIZE IS 0");
				return null;
			}
			
			// first check if we're really far behind and need to discard
			// samples at the head. if so, discard them.
			
			long lWaitTime;
			
			while ((lWaitTime = (iBufferLatency - ((int)System.currentTimeMillis() - ( snHead.later.lPacketRecordTime + iDeltaTime) )))
				   < 0  && iSize > 0)
			{
				// clear pointers so garbage collecter will find it
				SampleNode snTemp = snHead.later;
				snHead.later.later.earlier = snHead;
				snHead.later = snHead.later.later;
				
				//				System.out.println("REMOVED SAMPLE: " + snTemp.iSequenceNumber);

				snTemp.later = null;
				snTemp = null;
				
				iSize --;
				iRemovedSamples++;
			}
			// FIXME: should return silence packet.
			if (iSize == 0)
			{
				iNullSamples ++;
				return null;
			}
			
			// now check whether we're early ; if so, wait 
			
			// FIXME: Look at these constants....
			if (lWaitTime > 30)
			{
				try
				{
					snHead.wait((int)lWaitTime -15);
				}
				catch (InterruptedException e)
				{
					e.printStackTrace();
				}
			}
			
			iLastPacket = snHead.later.iSequenceNumber;

			iSize --;
			bData = snHead.later.bData;
			
			SampleNode snTemp = snHead.later;
			
			snHead.later.later.earlier = snHead;
			snHead.later = snHead.later.later;


			snTemp.later = null;
			snTemp = null;
		}
		return bData;
	}

	public Stats getStatistics()
	{
		Stats s = new Stats();
		s.addStat("<BufferQueue " + iMyInstance + "> Number of samples in buffer ", new Integer(iSize));
		s.addStat("<BufferQueue " + iMyInstance + "> Null samples delivered ", new Integer(iNullSamples));
		s.addStat("<BufferQueue " + iMyInstance + "> Removed samples ", new Integer(iRemovedSamples));
		s.addStat("<BufferQueue " + iMyInstance + "> Time Delta ", new Integer(iDeltaTime));
		s.addStat("<BufferQueue " + iMyInstance + "> Old Packets Received ", new Integer(iNumOldPacketsReceived));
		s.addStat("<BufferQueue " + iMyInstance + "> Reordered Packets ", new Integer(iReordered));
		
		return s;
	}
	

	public void dump(OutputStream os)
	{
		PrintWriter pw = new PrintWriter(os);
		pw.println("*** BufferQueue Size: " + iSize);
		SampleNode snTraversal = snHead;
		while (snTraversal != null)
		{
			if (snTraversal.iSequenceNumber == -1)
			{
				pw.println("   Head");
				
			}
			else if (snTraversal.iSequenceNumber == -2)
			{
				pw.println("   Tail");
			}
			else
			{
				pw.println("   Sequence: " + snTraversal.iSequenceNumber + 
						   "; (Creation) TimeStamp: " + snTraversal.lPacketRecordTime +
						   ( snTraversal.bIsSilence ? " <SILENCE> " : ""));
				
			}
			snTraversal = snTraversal.later;
			
		}
		pw.flush();
		
	}
}
