package jp.hasc.hasctool.core.runtime.filter.interpolator;

import java.util.ArrayList;

import jp.hasc.hasctool.core.data.SignalMessage;
import jp.hasc.hasctool.core.messaging.MessageConnector;
import jp.hasc.hasctool.core.messaging.MessageProcessor;
import jp.hasc.hasctool.core.messaging.MessageQueue;
import jp.hasc.hasctool.core.runtime.AbstractTask;
import jp.hasc.hasctool.core.runtime.RuntimeContext;

/**
 * 複数の入力ポートから信号を入力するフィルタのための抽象クラスです。
 * 複数の入力ポートからの信号（SignalMessage系列）の時刻を同期し、processSignalMessagesメソッドを呼び出します。
 * 時刻系列は、プライマリ入力ポート（setPrimaryPortIndexで指定）の時刻系列に合わせます。
 * 入力ポートの数を、setInputPortCountで指定して下さい。
 * @author iwasaki
 */
public abstract class AbstractMultipleInputsFilter extends AbstractTask {
	
	// それぞれの入力はキューに積まれ、それを別スレッドから取り出して処理される
	protected ArrayList<MessageQueue> inputPorts_=new ArrayList<MessageQueue>();
	
	protected ArrayList<SignalInterpolator> interpolators_=new ArrayList<SignalInterpolator>();
	
	protected int primaryInputPortIndex_;
	
	protected int inputPortCount_=0;
	
	private MessageConnector outputPort_ = new MessageConnector();
	
	public MessageConnector getOutputPort() { return outputPort_; }
	
	protected void outputMessage(Object message) throws InterruptedException {
		getOutputPort().processMessage(message);
	}
		
	public int getInputPortCount() {
		return inputPortCount_;
	}
	
	/**
	 * 入力ポートの数を指定します。
	 */
	public void setInputPortCount(int size) {
		inputPortCount_=size;
	}
	
	/**
	 * 指定したインデックスの入力ポートを返します。
	 */
	public MessageProcessor getInputPort(int index) {
		return inputPorts_.get(index);
	}
	
	public int getInputPortIndex(MessageProcessor idx) {
		return inputPorts_.indexOf(idx);
	}
	
	public abstract SignalInterpolator createSignalInterpolator(int portIndex);
	
	public MessageQueue createMessageQueue(int portIndex) {
		return getRuntimeContext().createDefaultMessageQueue();
	}
	
	@Override
	public void setup(RuntimeContext context) {
		super.setup(context);
		
		// create input ports
		interpolators_.clear(); inputPorts_.clear();
		for(int i=0;i<inputPortCount_;++i) {
			interpolators_.add(createSignalInterpolator(i));
			inputPorts_.add(createMessageQueue(i));
		}
	}

	public int getPrimaryInputPortIndex() {
		return primaryInputPortIndex_;
	}
	
	/**
	 * プライマリ入力ポート（時刻同期に使う信号を入力するポート）のインデックスを指定します。
	 */
	public void setPrimaryInputPortIndex(int timerPortIndex) {
		primaryInputPortIndex_ = timerPortIndex;
	}  

	@Override
	protected void run() throws InterruptedException {
		int psize = getInputPortCount();
		SignalMessage[] current = new SignalMessage[psize];
		
		final int ST_MAIN=1, ST_ENDED=2;
		int[] portState = new int[psize];
		for(int i=0;i<psize;++i) portState[i]=ST_MAIN;
		//
		outputMessage(SignalMessage.BEGIN);
		loop: while(!isShutdown()) {
			
			// primary port
			int pi = primaryInputPortIndex_;
			MessageQueue p_port = inputPorts_.get(pi);
			Object p_obj = p_port.readMessage(); // may be blocked
			if (p_obj instanceof SignalMessage) {
				SignalMessage p_sig = (SignalMessage) p_obj;
				SignalInterpolator p_itp = interpolators_.get(pi);
				if (!acceptSignalMessage(pi,p_sig) || !p_itp.addSample(p_sig)) {
					processUnacceptedSignalMessage(pi,p_sig);
					continue;
				}
				current[pi]=p_sig;
				SignalMessage[] result = new SignalMessage[psize];
				long p_time = p_sig.getTime();
				result[pi]=p_itp.getAtTime(p_time);
				
				// iterate other ports
				for(int i=0;i<psize;++i) {
					if (i==pi) continue;
					SignalInterpolator i_itp = interpolators_.get(i);
				 if (portState[i]==ST_ENDED) {
						result[i]=i_itp.getAtTime(p_time);
						continue;
					}
					MessageQueue i_port = inputPorts_.get(i);
					SignalMessage i_cursig=current[i];
					// synchronize
					while(i_cursig==null || i_cursig.getTime()<p_time) {
						if (isShutdown()) break loop;
						Object i_obj = i_port.readMessage(); // may be blocked
						if (i_obj instanceof SignalMessage) {
							SignalMessage i_sig = (SignalMessage) i_obj;
							if (!acceptSignalMessage(i,i_sig) || !i_itp.addSample(i_sig)) {
								processUnacceptedSignalMessage(i,i_sig);
								continue;
							}
							i_cursig=i_sig;
						} else if (i_obj==SignalMessage.BEGIN){
							// NOP
						} else if (i_obj==SignalMessage.END){
							portState[i]=ST_ENDED;
							break; // このポートのみをEND
							//break loop; // どれか一つのポートがENDしたら、全体をENDに？
						}else{
							processOtherMessage(i,i_obj);
						}
					}
					current[i]=i_cursig;
					result[i]=i_itp.getAtTime(p_time);
				}
				// output
				processSignalMessages(p_time, result);
				
			} else if (p_obj==SignalMessage.BEGIN){
				// NOP
			} else if (p_obj==SignalMessage.END){
				portState[pi]=ST_ENDED;
				break loop;
			}else {
				processOtherMessage(pi,p_obj);
			}
		}
		outputMessage(SignalMessage.END);
	}
	
	protected void processOtherMessage(int portIndex, Object message) throws InterruptedException {
		// primaryInputPortの場合のみ、そのまま出力
		if (portIndex==primaryInputPortIndex_) outputMessage(message);
	}

	protected void processUnacceptedSignalMessage(int portIndex, SignalMessage signalMessage) throws InterruptedException {
		// 処理しなかったSignalMessageは、無視する
	}

	/**
	 * @return SignalMessageを処理するならtrue、無視してスルーするならfalseを返します。
	 */
	protected boolean acceptSignalMessage(int portIndex, SignalMessage signalMessage) {
		return true;
	}

	/**
	 * 時刻同期された、SignalMessage配列を処理します。
	 * @throws InterruptedException 
	 */
	protected abstract void processSignalMessages(long time, SignalMessage[] messages) throws InterruptedException;
}
