/* $Id: TspSession.java,v 1.6 2006/06/27 20:46:42 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.consumer;

import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.ArrayList;

import tsp.core.common.TspAnswerExtendedInfo;
import tsp.core.common.TspAnswerOpen;
import tsp.core.common.TspAnswerSample;
import tsp.core.common.TspAnswerSampleFinalize;
import tsp.core.common.TspAnswerSampleInit;
import tsp.core.common.TspCommonException;
import tsp.core.common.TspDataInputStream;
import tsp.core.common.TspGroup;
import tsp.core.common.TspGroupItem;
import tsp.core.common.TspRequestExtendedInfos;
import tsp.core.common.TspRequestInfos;
import tsp.core.common.TspRequestSample;
import tsp.core.common.TspSampleFifo;
import tsp.core.common.TspSampleSet;
import tsp.core.common.TspSampleSymbols;
import tsp.core.common.TspStreamReader;
import tsp.core.common.url.TspURL;
import tsp.core.common.url.TspURLException;
import tsp.core.common.url.TspUnknownProtocolException;
import tsp.core.config.TspConfig;
import tsp.core.rpc.TSP_answer_extended_information_t;
import tsp.core.rpc.TSP_answer_open_t;
import tsp.core.rpc.TSP_answer_sample_destroy_t;
import tsp.core.rpc.TSP_answer_sample_init_t;
import tsp.core.rpc.TSP_answer_sample_t;
import tsp.core.rpc.TSP_argv_item_t;
import tsp.core.rpc.TSP_argv_t;
import tsp.core.rpc.TSP_request_open_t;
import tsp.core.rpc.TSP_request_sample_destroy_t;
import tsp.core.rpc.TSP_request_sample_init_t;
import tsp.core.rpc.TSP_sample_symbol_info_t;
import tsp.core.rpc.TSP_status_t;
import tsp.core.rpc.TspCommandChannelException;
import tsp.core.rpc.TspRequestSender;

/**
 * The TSP Session class represents a TSP session
 * between a TSP consumer and a TSP provider.
 * Since TSP is n to n protocol. A TSP Consumer may handle several sessions.
 * A session object contains all the information needed to operate a Session.
 */
public class TspSession {

	/**
	 * The session status describes the different
	 * state of a TSP session.
	 * SAMPLE_FINALIZE = sampling process is being stopped.
	 */
	public interface TspSessionStatus {
		/**
		 * Any kind of error occured,
		 * the session may be unusable
		 */
		public static final int BROKEN = -10;
		/**
		 * Default state of a newly created session.
		 */
		public static final int IDLE = 0;
		/**
		 * The TSP request sender has been 
		 * properly created/initialized and 
		 * is ready to send TSP request.
		 */
		public static final int SENDER_READY = 10;
		/**
		 * The TSP request Open has been sent
		 * successfully the answerOpen member is 
		 * meaningfull.
		 */
		public static final int OPENED = 20;
		/**
		 * One or several TSP requests sample have been
		 * sent successfully but the sampling
		 * process has not been initialized.
		 */
		public static final int SAMPLE_CONF = 30;
		/**
		 * The TSP request sample init request has
		 * been sent, the session should go automatically
		 * to the sample receive state
		 */
		public static final int SAMPLE_INIT = 40;
		/**
		 * Sample symbols are beeing received
		 * by the TspDataStreamReader 
		 * from the provider, TspDataStreamWriter.
		 */
		public static final int SAMPLE_RECEIVE = 50;
		/**
		 * Sampling process is being stopped.
		 * The TSP session should go automatically
		 * to the SAMPLE_CONF state (TBC).
		 */
		public static final int SAMPLE_FINALIZE = 60;
	}

	public interface TspSessionListener {
		void closeSession();
	}
	
	/**
	 * The Session state
	 * (Opened, SampleReceive, SampleFinalize...)
	 */
	private int state;
	
	private int lastTspStatus;
	
	/**
	 * The Session TSP URL.
	 */
	private TspURL url;
	
	/**
	 * Listeners
	 */
	private ArrayList sessionListeners; 

	/**
	 * the TSP answer sample obtained using
	 * the TSP request infos
	 */
	private TspAnswerSample answerInfos;
	
	/**
	 * the TSP answer extended info obtained using
	 * the TSP request extended info
	 */
	private TspAnswerExtendedInfo answerExtendedInfos;

	/**
	 * the TSP answer sample obtained using
	 * the TSP request sample
	 */
	private TspAnswerSample answerSample;	
	
	/**
	 * The socket used to read the TSP sample stream.
	 */
	private Socket tspSocket;

	/**
	 * The TSP stream Reader.
	 */
	private TspStreamReader tspStreamReader;
	
	/**
	 * The sample set storing sample read by
	 * the sample thread when used.
	 * Note that the use of the sample set and thread
	 * in the session is nw optionnal.
	 */
	private TspSampleSet sampleSet;
	
	/**
	 * The Sample FIFO thread
	 */
	private Thread sampleThread;
	
	/**
	 * The TSP request handler of the session
	 */
	private TspRequestSender requestSender;

	/**
	 * The session identifier obtained after
	 * a successfull TSP request open has been
	 * sent by the request sender. And is a part of 
	 * answerOpen object.
	 * The session id is the channel id of the TSP
	 * protocol which is used in subsequent TSP request.
	 */
	private TspAnswerOpen answerOpen;

	/**
	 * The description of the groups
	 */
	public TspGroup[] groups;
	
	/**
	 * Construct an empty TspSession
	 */
	public TspSession() {

		this.state            = TspSessionStatus.IDLE;
		this.lastTspStatus    = TSP_status_t.TSP_STATUS_OK;
		this.url              = null;
		this.sessionListeners = new ArrayList();
		
		this.answerInfos         = null;
		this.answerExtendedInfos = null;
		this.answerSample        = null;
		this.tspSocket           = null;
		this.tspStreamReader     = null;
		this.sampleSet           = null;
		this.sampleThread        = null;	
		this.requestSender       = null;
		this.answerOpen          = null;
		this.groups              = null;
	}
	
	/**
	 * Private Helper Method for verifying session state.
	 * Session is verified to be greater or equal to atLeastState 
	 * and no strictly greater than noMoreState.
	 * @param atLeastState    lower bound of state range (included)
	 * @param noMoreThanState upper bound of state range (included)
	 * @throws TspConsumerException 
	 */
	private void requiresStateRange(int atLeastState, int noMoreThanState) throws TspConsumerException {
	   if ((this.state < atLeastState) ||
		   (this.state >  noMoreThanState)
		   ) {
		   throw new TspConsumerException("Invalid Session State expected ["+atLeastState+","+noMoreThanState+"] but found <"+this.state+">");
	   }
	}
	
	public void register(TspSessionListener listener) {
		sessionListeners.add(listener);
	}
	
	public void unRegister(TspSessionListener listener) {
		sessionListeners.remove(listener);
	}	
	
	public int getChannelId() throws TspConsumerException {
		requiresStateRange(TspSessionStatus.OPENED,TspSessionStatus.SAMPLE_RECEIVE);
		return answerOpen.theAnswer.channel_id;
	}
	
	public int getTspVersionId() throws TspConsumerException {
		requiresStateRange(TspSessionStatus.OPENED,TspSessionStatus.SAMPLE_RECEIVE);
		return answerOpen.theAnswer.version_id;
	}

	/**
	 * The state of the TSP session.
	 * @return the state of the session.
	 */
	public int getState() {
		return state;
	}

	/**
	 * Open a TSP session between a consumer
	 * and a TSP provider.
	 * @param hostname the FQN host name where the TSP provider is up and running.
	 * @param tspProgramId the TSP provider Id. A TSP provider has an associated
	 *        Id in order to be able to run several provider on the same host.
	 * @throws TspURLException
	 * @throws TspConsumerException
	 */
	public int open(TspURL url) throws TspURLException, TspConsumerException {

		try {
			
			requiresStateRange(TspSessionStatus.IDLE,TspSessionStatus.IDLE);
			this.url = url;
			/* TODO : create a factory depending on the protocol in order to obtain a request sender */
			if(url.getProtocol() == null){
				throw new TspUnknownProtocolException("No protocol specified");
			}
			if(!url.getProtocol().equals("rpc")){
				throw new TspUnknownProtocolException("Unknown protocol : "  + url.getProtocol());
			}
											
			requestSender = new TspRequestSender(url);
								
			state = TspSessionStatus.SENDER_READY;
			/* Build the TSP request open */
			TSP_request_open_t reqO = new TSP_request_open_t();
			reqO.version_id = TspConfig.PROTOCOL_VERSION;
			reqO.argv = new TSP_argv_t();
			reqO.argv.value = null;
			reqO.argv.value = new TSP_argv_item_t[0];
			/* Send the request open to RPC server to open the session */
			TSP_answer_open_t ansO = new TSP_answer_open_t();

			ansO = requestSender.open(reqO);

			answerOpen = new TspAnswerOpen(ansO);
			lastTspStatus = answerOpen.getTspStatus();

			state = TspSessionStatus.OPENED;
			return answerOpen.theAnswer.channel_id;		
		} catch (TspCommandChannelException e) {
			throw new TspConsumerException(e);
		}
	}

	/**
	 * Close the TSP session.
	 */
	public void close() throws TspConsumerException {

		requiresStateRange(TspSessionStatus.OPENED,TspSessionStatus.SAMPLE_FINALIZE);
				
		if (state == TspSessionStatus.SAMPLE_RECEIVE) {
			requestSampleFinalize();
		}
		/* FIXME stop data receiving thread properly */
		state = TspSessionStatus.IDLE;
	}

	/**
	 * Returns the scalar size of the requested samples
	 * by adding each of the samples dimension
	 * @return the scalar size of the requested samples
	 */
	public int getScalarSize() {
		int retval = 0;
		for(int i=0;i<answerSample.theAnswer.symbols.value.length;i++){
			retval += answerSample.theAnswer.symbols.value[i].dimension;
		}
		return retval;	
	}
	
	/**
	 * Send request infos to provider if necessary.
	 * @return the AnswerSample.
	 */
	public TspAnswerSample requestInfos(TspRequestInfos req) throws TspConsumerException {

		requiresStateRange(TspSessionStatus.OPENED,TspSessionStatus.SAMPLE_CONF);
		try {
			if (answerInfos == null) {
				req.theRequest.channel_id = answerOpen.theAnswer.channel_id;
				TSP_answer_sample_t theAnswer = requestSender.information(req.theRequest);
				answerInfos = new TspAnswerSample(theAnswer);
				lastTspStatus = answerInfos.getTspStatus();
			}
			return answerInfos;
		}
		catch (TspCommandChannelException e) {
			throw new TspConsumerException(e);
		}
	}
	
	public  TspAnswerSample requestInfos() throws TspConsumerException {
		TspRequestInfos  reqI = new TspRequestInfos(getTspVersionId(),getTspVersionId());
		return requestInfos(reqI);
	}
	
	/**
	 * Send request extended infos to provider if necessary.
	 * @return the AnswerExtendedInfo.
	 */
	public TspAnswerExtendedInfo requestExtendedInfos(TspRequestExtendedInfos req) throws TspConsumerException {

		requiresStateRange(TspSessionStatus.OPENED,TspSessionStatus.SAMPLE_CONF);
		try {
			TSP_answer_extended_information_t theAnswerExtended = requestSender.information(req.theRequest);
			answerExtendedInfos = new TspAnswerExtendedInfo(theAnswerExtended);
			lastTspStatus = answerExtendedInfos.getTspStatus();
			return answerExtendedInfos;
		}
		catch (TspCommandChannelException e) {
			throw new TspConsumerException(e);
		}
	}



	/**
	 * Send a TSP request sample on an opened session.
	 * @req the TSP request Sample
	 */
	public TspAnswerSample requestSample(TspRequestSample req) throws TspConsumerException, TspConsumerRequestSampleException {

		requiresStateRange(TspSessionStatus.OPENED,TspSessionStatus.SAMPLE_CONF);
		try {
			TSP_answer_sample_t theAnswer = new TSP_answer_sample_t();
			req.theRequest.channel_id = answerOpen.theAnswer.channel_id;
			theAnswer = requestSender.sample(req.theRequest);
			
			

			state = TspSessionStatus.SAMPLE_CONF;
			/* We cache the result for future use */
			answerSample = new TspAnswerSample(theAnswer);
			lastTspStatus = answerSample.getTspStatus();
			
			/* Verify the session state */
			if (answerSample.theAnswer.status != TSP_status_t.TSP_STATUS_OK){
				throw new TspConsumerRequestSampleException(answerSample.theAnswer);
			}

			TspConfig.log(TspConfig.LOG_FINE, "answerSample");
			/* build the groups */
			buildSampleGroups();
			return answerSample;
		}
		catch (TspCommandChannelException e) {
			throw new TspConsumerException(e);
		}

	}

	/**
	 * Send a TSP request sample Init on an opened session.
	 * This ask for beginning the sample process and 
	 * to open the socket for receiving sample.
	 */
	public TspAnswerSampleInit requestSampleInit() throws TspConsumerException {

		sampleSet = new TspSampleFifo(TspConfig.TSP_FIFO_SIZE);
		TspConfig.log(TspConfig.LOG_FINE, "Sample FIFO created");
		
		return requestSampleInit(sampleSet);
	}
	
	/**
	 * Send a TSP request sample Init on an opened session.
	 * This ask for beginning the sample process and 
	 * to open the socket for receiving sample.
	 */
	public TspAnswerSampleInit requestSampleInit(TspSampleSet sampleSet_) throws TspConsumerException {

		requiresStateRange(TspSessionStatus.SAMPLE_CONF,TspSessionStatus.SAMPLE_CONF);
		try {
			TSP_request_sample_init_t reqSI = new TSP_request_sample_init_t();
			TSP_answer_sample_init_t theAnswer = new TSP_answer_sample_init_t();
			reqSI.version_id = answerOpen.theAnswer.version_id;
			reqSI.channel_id = answerOpen.theAnswer.channel_id;

			theAnswer = requestSender.sampleInit(reqSI);

			state = TspSessionStatus.SAMPLE_INIT;
			TspAnswerSampleInit asi = new TspAnswerSampleInit(theAnswer);
			lastTspStatus = asi.getTspStatus();
			
			/* create the Sample Fifo */
			/*Modif : Mathias Choquet Mai 2005. 
			 * In order to make the tspSampleFifo Optionnal.
			 */
			sampleSet = sampleSet_;
			TspConfig.log(TspConfig.LOG_FINE, "Sample FIFO created");
			
			/* Open, socket using answer sample */
			tspSocket = createTspSocket(asi);
			
			/* create the TspStreamReader */
			tspStreamReader = new TspStreamReader(new TspDataInputStream(tspSocket), groups, sampleSet);
			
			/* create Thread */
			sampleThread = new Thread(tspStreamReader);
			/* run it */
			sampleThread.start();

			return asi;
		}
		catch (TspCommandChannelException e) {
			throw new TspConsumerException(e);
		}
		catch (TspCommonException e) {
			throw new TspConsumerException(e);
		}
		catch (UnknownHostException e) {
			throw new TspConsumerException(e);
		}
	}	
	
	/**
	 * 
	 *
	 */
	public TspAnswerSampleFinalize requestSampleFinalize() throws TspConsumerException {

		requiresStateRange(TspSessionStatus.SAMPLE_INIT,TspSessionStatus.SAMPLE_RECEIVE);
		try {
			TSP_request_sample_destroy_t reqSF = new TSP_request_sample_destroy_t();
			TSP_answer_sample_destroy_t theAnswer = new TSP_answer_sample_destroy_t();
			reqSF.version_id = answerOpen.theAnswer.version_id;
			reqSF.channel_id = answerOpen.theAnswer.channel_id;

			theAnswer = requestSender.sampleFinalize(reqSF);

			state = TspSessionStatus.SAMPLE_CONF;

			/* stop the stream reader thread properly */
			tspStreamReader.stopMe();
			/* wait for the thread death */
			try {
				sampleThread.join(1000);
			}
			catch (Exception e) {
			} /* FIXME que faire quand un thread est interrompu?*/
			/* close the socket */

			tspSocket.close();
			lastTspStatus = theAnswer.status;
			return new TspAnswerSampleFinalize(theAnswer);
		}
		catch (TspCommandChannelException e) {
			throw new TspConsumerException(e);
		}
		catch (IOException e) {
			throw new TspConsumerException(e);
		}
	}
		

	/**
	 * The sample set storing received sample
	 * when needed.
	 * All the method of the class implementing the FIFO
	 * are synchronized since the Fifo is read (get) by the
	 * final TSP consumer application and written by 
	 * the Tsp Stream Reader Thread.
	 */	
	public TspSampleSet getSampleSet() {
		return sampleSet;
	}

	/**
	 * Build the sample group with  the last answerSample
	 */
	private void buildSampleGroups() throws TspConsumerException{
				
		
		if (answerSample != null) {
			/* Create as many groups as specified by the 
			 * provider in the answer sample
			 */
			groups = new TspGroup[answerSample.theAnswer.provider_group_number];
			for (int j = 0; j < groups.length; ++j) {
				groups[j] = new TspGroup();
			}
			/*
			 * Build simple Sample Symbol interface 
			 */
			TspSampleSymbols tspss = new TspSampleSymbols(answerSample);
			/*
			 * Loop over symbols 
			 */
			TspConfig.log(
				TspConfig.LOG_FINE,
				"TspSession::buildSampleGroups:" + "Building sample groups for <" + tspss.nbSymbols() + "> symbols in <" + groups.length + "> groups.");
			TSP_sample_symbol_info_t s;
			TspGroupItem item;

			for (int i = 0; i < tspss.nbSymbols(); ++i) {
				
				/* retrieve symbol infos */
				s = (TSP_sample_symbol_info_t) tspss.getSymbol(i);	
				
				/* create group item */
				/*item = new TspGroupItem(s.provider_global_index, 
							tsp.core.config.TspConfig.SIZE_OF_ENCODED_DOUBLE, 
							s.provider_group_rank);*/
				
				item = new TspGroupItem(s);
				

				/* add to appropriate group */
				groups[s.provider_group_index].addItem(item);
			} /* end loop over symbols */
			/* Sort the groups:
			 * i.e. each item of one group should appear in the 
			 * lowest group rank first.
			 */
			for (int j = 0; j < groups.length; ++j) {
				groups[j].sortGroup();
			}
		} /* end if answerSample != null */
		/** FIXME should raise exception but should NEVER happen
		 ** since its a private method.
		 **/
	}

	/**
	 * Create (and connect) 
	 * the TSP socket using the answer sample init infos.
	 */
	private Socket createTspSocket(TspAnswerSampleInit asi) throws UnknownHostException, TspConsumerException {

		//JDK 1.4 String [] ap = asi.theAnswer.data_address.split(":");

		String[] ap = new String[2];
		int iodp = asi.theAnswer.data_address.indexOf(':');
		ap[0] = asi.theAnswer.data_address.substring(0, iodp);
		ap[1] = asi.theAnswer.data_address.substring(iodp + 1);

		Socket sock = null;

		try {
			sock = new Socket(InetAddress.getByName(ap[0]), (new Integer(ap[1])).intValue());
			/* Configure socket */
			sock.setReceiveBufferSize(tsp.core.config.TspConfig.DATA_STREAM_SOCKET_FIFO_SIZE / 2);
			//sock.setReuseAddress(true); FIXME JDK.1.2.2
			//sock.setKeepAlive(false);  FIXME JDK.1.2.2
			sock.setTcpNoDelay(true);
			/* 50 ms timeout on read 
			 * FIXME how to set that ???
			 */
			sock.setSoTimeout(500);
			TspConfig.log(TspConfig.LOG_INFO, "Socket connected.");

			return sock;
		}
		catch (NumberFormatException e) {
			throw new TspConsumerException(e);
		}
		catch (SocketException e) {
			throw new TspConsumerException(e);
		}
		catch (IOException e) {
			throw new TspConsumerException(e);
		}

	}

	public int getLastTspStatus() {
		return lastTspStatus;
	}
	
	

}

// End of TspSession.java
