package jp.hasc.hasctool.core.runtime.filter.file.path;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;

import jp.hasc.hasctool.core.runtime.RuntimeContext;
import jp.hasc.hasctool.core.runtime.filter.AbstractFilter;
import jp.hasc.hasctool.core.util.CoreUtil;

/**
 * File Reader
 * input:filePath
 * output:(filePath,)val1,val2,....
 * @author hiro
 */

public class FileReader extends AbstractFilter{

	boolean addFilePath_ = false;

	@Override
	public void processMessage(Object message) throws InterruptedException {
		if (message instanceof String) {
			String filePath = (String)message;
			System.out.println("now reading : " + filePath);
			try{
				InputStream inps = getRuntimeContext().getFileStreamProvider().openInputStream(filePath);
				BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inps, RuntimeContext.DEFAULT_CHARSET));
				while(true){
					String line = bufferedReader.readLine();
					if (line == null) {
						break;
					} else {
						if(addFilePath_){
							outputMessage(filePath + "," + line);
						} else {
							outputMessage(line);
						}
					}
				}
				bufferedReader.close();
				inps.close();
			} catch (IOException ex) {
				CoreUtil.throwAsRuntimeException(ex);
			} catch (RuntimeException ex) {
			}
		}else{
			//BEGIN,END message
			outputMessage(message);
		}
	}

	public void setAddFilePath(boolean addFilePath){
		addFilePath_ = addFilePath;
	}

}
