package jp.hasc.hasctool.core.runtime.filter.weka.message;

import java.util.ArrayList;
import jp.hasc.hasctool.core.data.EnumCommand;
import jp.hasc.hasctool.core.messaging.MessageProcessor;
import jp.hasc.hasctool.core.messaging.MessageQueue;
import jp.hasc.hasctool.core.runtime.RuntimeContext;
import jp.hasc.hasctool.core.runtime.source.AbstractSource;

/**
 * 複数の入力ポートから入力されたmessageを、入力ポート順に出力します。
 * 例えば、ポート0から BEGIN,A0,B0,END、ポート1から BEGIN,A1,B1,END 
 * という系列が入力された場合、BEGIN,A0,B0,A1,B1,END　という系列を出力します。
 * @author hiro
 */

public class MessageMerger extends AbstractSource {
	// それぞれの入力はキューに積まれ、それを別スレッドから取り出して処理される
	protected ArrayList<MessageQueue> inputPorts_=new ArrayList<MessageQueue>();

	private int inputPortCount_;

	public int getInputPortCount() {
		return inputPortCount_;
	}

	/**
	 * 入力ポートの数を指定します。
	 */
	public void setInputPortCount(int size) {
		inputPortCount_=size;
	}

	/**
	 * 指定したインデックスの入力ポートを返します。
	 */
	public MessageProcessor getInputPort(int index) {
		return inputPorts_.get(index);
	}

	public MessageQueue createMessageQueue(int portIndex) {
		return getRuntimeContext().createDefaultMessageQueue();
	}

	@Override
	public void setup(RuntimeContext context) {
		super.setup(context);

		// create input ports
		inputPorts_.clear();
		for(int i=0;i<inputPortCount_;++i) {
			inputPorts_.add(createMessageQueue(i));
		}
	}

	@Override
	protected void run() throws InterruptedException {
		int psize = getInputPortCount();
		outputMessage(EnumCommand.BEGIN);

		//入力ポート順にmessageを送っています
		for(int i=0;i<psize;++i) {
			while(true){
				Object message = inputPorts_.get(i).readMessage();
				if (message==EnumCommand.END){
					break;
				} else if (message!=EnumCommand.BEGIN){
					outputMessage(message);
				}
			}
		}
		//
		outputMessage(EnumCommand.END);
	}
}
