// Path: src/main/java/com/ilog/util/security/PropertyUtil.java
// public class PropertyUtil {
// 	private static Logger logger = Logger.getLogger(PropertyUtil.class);
// 	
// 	/**
// 	 * 获取当前项目配置文件根路径 
// 	 * @return java.lang.String
// 	 */
// 
// 	public static String getCurrentConfPath() {
// 		String c_path = PropertyUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath();
// 		String os_name = System.getProperty("os.name").toLowerCase();
// 		c_path = os_name.startsWith("win") ? c_path.substring(1, c_path.lastIndexOf("/") + 1) : c_path.substring(0, c_path.lastIndexOf("/") + 1);
// 		return c_path + "../conf/";
// 	}
// }


import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.apache.log4j.PropertyConfigurator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ilog.util.security.PropertyUtil;

		if(headers != null){
			eventList.add(EventBuilder.withBody(body, headers));
		}
		if (eventList.size()>0 && (eventList.size() == batchUpperLimit || timeInterval>=TIME_INTERVAL)) {//批量提交Event
			getChannelProcessor().processEventBatch(eventList);
			eventList.clear();
			startTime = cur;
		}
	}

	public synchronized void stop() {
		log.info("File Tailer Source is stopping.");
		//确保读取信息写入磁盘文件不丢失
		tailer.stop();
		//实时采集主题程序
		if(log.isInfoEnabled()){
			log.info("file tailer is stopped,wait for bakup marks while sleeping 5s.");
		}
		try{
			Thread.sleep(5000);// 为了让下一行fReader.readLine()读取不会出现null的情况
		}catch(InterruptedException ex){
			log.error("File Tailer Source is intterrupted while sleeping.", ex);
		}
		this.handle(null, new HashMap<String,String>(){{put(FileTailer.FLUME_STOP_TIMER_KEY,null);}});//保证event及时刷新
		super.stop();
		log.info("File Tailer Source is stopped.");
	}

	public void configure(Context context) {
		batchUpperLimit = context.getInteger("batchUpperLimit",1);

// Path: src/main/java/com/ilog/util/security/PropertyUtil.java
// public class PropertyUtil {
// 	private static Logger logger = Logger.getLogger(PropertyUtil.class);
// 	
// 	/**
// 	 * 获取当前项目配置文件根路径 
// 	 * @return java.lang.String
// 	 */
// 
// 	public static String getCurrentConfPath() {
// 		String c_path = PropertyUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath();
// 		String os_name = System.getProperty("os.name").toLowerCase();
// 		c_path = os_name.startsWith("win") ? c_path.substring(1, c_path.lastIndexOf("/") + 1) : c_path.substring(0, c_path.lastIndexOf("/") + 1);
// 		return c_path + "../conf/";
// 	}
// }


// Path: src/main/java/com/ilog/flume/source/tail/FileTailerSource.java
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.apache.log4j.PropertyConfigurator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ilog.util.security.PropertyUtil;

		if(headers != null){
			eventList.add(EventBuilder.withBody(body, headers));
		}
		if (eventList.size()>0 && (eventList.size() == batchUpperLimit || timeInterval>=TIME_INTERVAL)) {//批量提交Event
			getChannelProcessor().processEventBatch(eventList);
			eventList.clear();
			startTime = cur;
		}
	}

	public synchronized void stop() {
		log.info("File Tailer Source is stopping.");
		//确保读取信息写入磁盘文件不丢失
		tailer.stop();
		//实时采集主题程序
		if(log.isInfoEnabled()){
			log.info("file tailer is stopped,wait for bakup marks while sleeping 5s.");
		}
		try{
			Thread.sleep(5000);// 为了让下一行fReader.readLine()读取不会出现null的情况
		}catch(InterruptedException ex){
			log.error("File Tailer Source is intterrupted while sleeping.", ex);
		}
		this.handle(null, new HashMap<String,String>(){{put(FileTailer.FLUME_STOP_TIMER_KEY,null);}});//保证event及时刷新
		super.stop();
		log.info("File Tailer Source is stopped.");
	}

	public void configure(Context context) {
		batchUpperLimit = context.getInteger("batchUpperLimit",1);

		PropertyConfigurator.configure(PropertyUtil.getCurrentConfPath() + "log4j.properties");
