/* $Id: TspStreamReader.java,v 1.3 2006/06/04 08:33:45 erk Exp $
 * -----------------------------------------------------------------------
 * 
 * TSP Library - core components for a generic Transport Sampling Protocol.
 * 
 * Copyright (c) 2002 Yves DUFRENNE, Stephane GALLES, Eric NOULARD and Robert PAGNOT 
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 * 
 * -----------------------------------------------------------------------
 * 
 * Project    : TSP
 * SubProject : jTSP
 * Maintainer : tsp@astrium-space.com
 * Component  : Consumer
 * 
 * -----------------------------------------------------------------------
 * 
 * Purpose   : 
 * 
 * -----------------------------------------------------------------------
 */

package tsp.core.common;

import tsp.core.config.*;
import tsp.core.rpc.TSP_datatype_t;
import tsp.consumer.util.configuration.SpecException;
import tsp.consumer.util.configuration.VarTypeEnum;

/**
 * The TSP StreamReader class is able to read
 * a TSP stream on a {@link DataInput} using the specified 
 * {@link TspGroups}.
 */
public class TspStreamReader implements Runnable {

	private int ok = 0;
	
    public TspStreamReader(TspDataInputStream in, TspGroup[] groups, TspSampleSet sampleSet) {
	this.in     = in;
	this.groups = groups;
	/*Modif : Mathias Choquet Mai 2005. 
	 * In order to make the tspSampleFifo Optionnal.
	 */
	this.sampleSet   = sampleSet;
	stop        = false;}
	
	
    
    /**
     * Read all sample symbol belonging
     * to the specified TSP group
     * @param i the number of teh TSP group to be read
     */
    public void readGroup(int i)  throws TspCommonException, TspException
    {	
	TspGroup     g = groups[i];
	Object       value = null;
	TspGroupItem item;

	/* wait to have enough data to read all group item */	      
	while (!in.available(g.encoded_size())) {		
	    try {
	    Thread.sleep(10);		
	    }
	    catch (InterruptedException e) {
		TspConfig.log(TspConfig.LOG_INFO,
			      "TspStreamReader::readGroup::thread interrompu");
	    }
	}
	/* Loop over each item of the group */
	for (int rank=0; rank < g.nbItem(); ++rank) {
	    /* get item */
	    item = g.getItem(rank);
	    
	    /* handle multi-type TSP symbols decode */
	    
	    // Which is the type returned by the provider ?
	   	switch(item.getReturnedType()){
	   	case(TSP_datatype_t.TSP_TYPE_DOUBLE):
	   		for(int elem=0;elem<item.getNbElem();++elem){
	   			value = new Double(in.tspDecodeDouble());	   			
	   		    /* add sample to the FIFO */
	   		    /*Modif : Mathias Choquet Mai 2005. 
	   		     * In order to make the tspSampleFifo Optionnal.
	   		     */
	   		    sampleSet.putSample(new TspSample(time_stamp,item.provider_global_index,value,elem));
	   		   
	   		    /* statistics */
	   		    ++sample_count;
	   		}
   			break;
   		
	    case(TSP_datatype_t.TSP_TYPE_FLOAT):
	    	for(int elem=0;elem<item.getNbElem();++elem){
	    		value = new Float (in.tspDecodeFloat());	
	    	    sampleSet.putSample(new TspSample(time_stamp,item.provider_global_index,value,elem));
	    	    ++sample_count;
	    	}
	   		break;

    	case(TSP_datatype_t.TSP_TYPE_INT64):
	    case(TSP_datatype_t.TSP_TYPE_UINT64):
	    	for(int elem=0;elem<item.getNbElem();++elem){
	    		value = new Long(in.tspDecodeLong());
	    	    sampleSet.putSample(new TspSample(time_stamp,item.provider_global_index,value,elem));
	    	    ++sample_count;
	    	}
	    	break;
	    	
	   	case(TSP_datatype_t.TSP_TYPE_INT8):
	   	case(TSP_datatype_t.TSP_TYPE_INT16):
    	case(TSP_datatype_t.TSP_TYPE_INT32):	
	    case(TSP_datatype_t.TSP_TYPE_UINT8):
	    case(TSP_datatype_t.TSP_TYPE_UINT16):
	    case(TSP_datatype_t.TSP_TYPE_UINT32):
	    	for(int elem=0;elem<item.getNbElem();++elem){
	    		value = new Integer(in.tspDecodeInt());
	    	    sampleSet.putSample(new TspSample(time_stamp,item.provider_global_index,value,elem));
	    	    ++sample_count;
	    	}
	    	break;
	    	
	    case(TSP_datatype_t.TSP_TYPE_CHAR):
	    case(TSP_datatype_t.TSP_TYPE_UCHAR):
	    	for(int elem=0;elem<item.getNbElem();++elem){
	    		value = new Character(in.tspDecodeChar());;
	    		sampleSet.putSample(new TspSample(time_stamp,item.provider_global_index,value,elem));
	    		++sample_count;
	    	}
	    	break;
	    	
	    case(TSP_datatype_t.TSP_TYPE_RAW):
	    	for(int elem=0;elem<item.getNbElem();++elem){
	    		value = new Byte(in.tspDecodeRaw());
	    		sampleSet.putSample(new TspSample(time_stamp,item.provider_global_index,value,elem));
	    		++sample_count;
	    	}
	    	break;
	    		
	    	
	    default:
	    	throw new TspException("TSP data type not supported for decoding sample");
	    }
	}
	
}

    public void run() {
	try {
	    while (!stop) {
		
		/* we must be able to read 
		 * time_stame and group_index 
		 */    
		while (!in.available(8)) {		    
		    try {
		    Thread.sleep(10);
		    }
		    catch (InterruptedException e) {
			TspConfig.log(TspConfig.LOG_INFO,"TspStreamReader::run::thread interrompu");
		    }
		}	    
		time_stamp  = in.tspDecodeInt();
		group_index = in.tspDecodeInt();
		//System.out.println(System.currentTimeMillis());
		if (checkGroupIndex(group_index)) {
		    try {
		    	readGroup(group_index);
		    }
		    catch (TspException e){
		    	TspConfig.log(TspConfig.LOG_WARNING,"TspException: Unable to read group<"+e.getMessage()+">");
		    }
		}
	    } /* infinite while */
	}
	catch (TspCommonException e) {
	    TspConfig.log(TspConfig.LOG_INFO,
			  "End of TSP stream");
	}

	TspConfig.log(TspConfig.LOG_INFO, "Thread is finishing...");
	/* bye bye guy... */
    } /* end pf run */
   

    public void stopMe() {	
	stop = true;
    }

    public boolean isStopped() {
	return stop;
    }

    private boolean checkGroupIndex(int group_index) {
	/* FIXME should check for special group case */
	if (group_index<0) {
	    TspConfig.log(TspConfig.LOG_WARNING,"Negative group index = "+group_index);
	    return false;
	}
	return true;
    }

    private boolean       stop;
    private TspDataInputStream     in;
    protected TspGroup[]    groups;
    
    /*Modif : Mathias Choquet Mai 2005. 
     * In order to make the tspSampleFifo Optionnal.
     */
    private TspSampleSet sampleSet;
    
    private int           time_stamp;
    private int           group_index;
    /* for statistic purpose */
    private long          sample_count;

} /* end of TspStreamReader */
