package org.cocolab.inpro.sphinx.frontend;

import java.net.InetAddress;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;


import edu.cmu.sphinx.frontend.BaseDataProcessor;
import edu.cmu.sphinx.frontend.Data;
import edu.cmu.sphinx.frontend.DataEndSignal;
import edu.cmu.sphinx.frontend.DataProcessingException;
import edu.cmu.sphinx.frontend.DataProcessor;
import edu.cmu.sphinx.frontend.DataStartSignal;
import edu.cmu.sphinx.frontend.DoubleData;
import edu.cmu.sphinx.util.props.PropertyException;
import edu.cmu.sphinx.util.props.PropertySheet;
import edu.cmu.sphinx.util.props.PropertyType;
import edu.cmu.sphinx.util.props.Registry;
import gov.nist.jrtp.RtpErrorEvent;
import gov.nist.jrtp.RtpListener;
import gov.nist.jrtp.RtpPacket;
import gov.nist.jrtp.RtpPacketEvent;
import gov.nist.jrtp.RtpSession;
import gov.nist.jrtp.RtpStatusEvent;
import gov.nist.jrtp.RtpTimeoutEvent;

public class RtpRecvProcessor extends BaseDataProcessor {

	public final static String PROP_RTP_RECV_PORT = "recvPort";
	public final static String PROP_SAMPLING_RATE = "samplingRate";
	
	private int recvPort = 42000;
	private int samplingRate = 16000;
	
	// temporarily store used when RTP data flows in more quickly than can be processed
	private BlockingQueue<Data> q;
	
	// 
	private long sampleNumber;
	
	private void resetFrameState() {
        sampleNumber = 0;
		q.add(new DataStartSignal(sampleNumber));
	}

	/*
	 * this is called by the next data processor in the frontend
	 */
	@Override
	public Data getData() throws DataProcessingException{
		Data d = null;
		while (d == null) {
			try {
				d = q.take();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return d;
	}
	
	@Override
	public void initialize() {
		super.initialize();
        q = new LinkedBlockingQueue<Data>();
        resetFrameState();
        try {
			RtpSession rs = new RtpSession(InetAddress.getLocalHost(), recvPort);
			rs.addRtpListener(new RtpHandler());
			rs.receiveRTPPackets();
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("Error setting up RtpSession in RtpRecvProcessor");
		}   
	}

	/*
	 * make sure, that nobody sets a predecessor to this data processor
	 */
	@Override
	public void setPredecessor(DataProcessor predecessor){
		if (predecessor != null) {
			throw new RuntimeException("can't set predecessor of an RtpRecvProcessor.");
		}
	}
	
	/* * configurable interface  **/
	
	@Override
	public void register(String name, Registry registry) throws PropertyException {
		super.register(name, registry);
		registry.register(PROP_RTP_RECV_PORT, PropertyType.INT);
		registry.register(PROP_SAMPLING_RATE, PropertyType.INT);
	}

	@Override
	public void newProperties(PropertySheet ps) throws PropertyException {
		super.newProperties(ps);
		recvPort = ps.getInt(PROP_RTP_RECV_PORT, recvPort);
		samplingRate = ps.getInt(PROP_SAMPLING_RATE, samplingRate);
	}

	/* * the handler for incoming RTP packets * */
	
	private class RtpHandler implements RtpListener {

	private void stop() {
		q.add(new DataEndSignal(sampleNumber));
		resetFrameState();
	}
	
	public void handleRtpPacketEvent(RtpPacketEvent e) {
		RtpPacket rp = e.getRtpPacket();
		byte[] bPayload = rp.getPayload();
		double[] dPayload = ConversionUtil.bytesToDoubles(bPayload);
		DoubleData data = new DoubleData(dPayload, samplingRate, 0, sampleNumber);
		q.add(data);
		sampleNumber += dPayload.length;
	}

	public void handleRtpErrorEvent(RtpErrorEvent e) {
		stop();
	}
	
	public void handleRtpStatusEvent(RtpStatusEvent e) {
		stop();
	}

	public void handleRtpTimeoutEvent(RtpTimeoutEvent e) {
		stop();
	}

	}
}

@@ -1146,43 +1146,8 @@
 00;%0A
-%09private int samplingRate = 16000;%0A
 %09%0A%09/
@@ -2616,67 +2616,8 @@
 T);%0A
-%09%09registry.register(PROP_SAMPLING_RATE, PropertyType.INT);%0A
 %09%7D%0A%0A
@@ -2784,70 +2784,8 @@
 t);%0A
-%09%09samplingRate = ps.getInt(PROP_SAMPLING_RATE, samplingRate);%0A
 %09%7D%0A%0A
@@ -3075,18 +3075,17 @@
 %09byte%5B%5D 
-bP
+p
 ayload =
@@ -3108,25 +3108,23 @@
 ;%0A%09%09
-d
+D
 ouble
-%5B%5D dPayload
+Data data
  = C
@@ -3154,12 +3154,14 @@
 uble
-s(bP
+Data(p
 aylo
@@ -3171,101 +3171,52 @@
 ;%0A%09%09
-DoubleData data = new DoubleData(dPayload, samplingRate, 0, sampleNumber);%0A%09%09q.add(data);%0A%09%09s
+q.add(data);%0A%09%09sampleNumber = data.getFirstS
 ampl
@@ -3226,20 +3226,29 @@
 mber
+()
  +
-=
  d
-Payload
+ata.getValues()
 .len

update RTP receiver to be compatible with the SimpleRTP as just checked in: