// Path: src/main/java/com/taodian/monitor/storm/DataBolt.java
// public interface DataBolt {
// 
// 	/**
// 	 * 准备开始处理数据。需要做一些数据初始化操作。
// 	 */
// 	public void prepare(TopologyContext context);
// 	
// 	
// 	/**
// 	 * 处理一个数据元素。
// 	 * 
// 	 * @param data
// 	 */
// 	public void execute(DataCell data, OutputCollector output);
// 
// 	/**
// 	 * 系统推出时调用。
// 	 */
// 	public void shutdown();
// 
// }
// 
// Path: src/main/java/com/taodian/monitor/storm/DataSpout.java
// public interface DataSpout {
// 	
// 	public void prepare(TopologyContext context);
// 	public DataCell nextDataCell();
// 	
// 	public boolean isClosed();
// }
// 
// Path: src/main/java/com/taodian/monitor/storm/Topology.java
// public interface Topology {
// 	
// 	/**
// 	 * 设置一个输入数据源，
// 	 * 
// 	 * @param name 输入数据源的名字。
// 	 * @param spout， 数据源
// 	 */
// 	public void setSpout(String name, DataSpout spout, String output);
// 
// 	public void setSpout(String name, DataSpout spout);
// 
// 	
// 	/**
// 	 * 设置一个输入数据源，
// 	 * 
// 	 * @param name 数据来源
// 	 * @param bolt，
// 	 * @param output -- 输出的消息队列。
// 	 */
// 	public void setBolt(String from, DataBolt bolt, String output);
// 	
// 	/**
// 	 * 开始运行
// 	 */
// 	public void start();
// 	
// 	public List<DataQueue> getAllDataQueue();
// }


import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.taodian.monitor.storm.DataBolt;
import com.taodian.monitor.storm.DataSpout;
import com.taodian.monitor.storm.ItemFactory;
import com.taodian.monitor.storm.Topology;
import com.taodian.monitor.storm.TopologyBuilder;
package com.taodian.monitor.storm.impl;




/**
 * 根据文本输入流配置拓扑。
 * @author deonwu
 */
public class SimpleTopologyBuilder implements TopologyBuilder{
	private Log log = LogFactory.getLog("storm.builder");  
	
	private BufferedReader reader = null;
	public SimpleTopologyBuilder(InputStream ins){
		try {
			reader = new BufferedReader(new InputStreamReader(ins, "utf8"));
		} catch (UnsupportedEncodingException e) {
			log.error(e, e);
		}
	}

	@Override
	public void buildToplogy(Topology topology, ItemFactory factory){
		try{
			log.info("-------Read topology configuration-------");
			for(String r = reader.readLine(); r != null; r = reader.readLine()){
				r = r.trim();
				if(r.startsWith("#") || r.length() == 0) continue;
				String[] tmp = r.split("=", 2);
				log.info(r);
				if(tmp.length == 2){
					String[] x = tmp[1].split(",", 2);
					Object obj = factory.create(x[0].trim());
					String p = x.length == 2 ? x[1] : "";
					if(obj instanceof DataSpout){
// Path: src/main/java/com/taodian/monitor/storm/DataBolt.java
// public interface DataBolt {
// 
// 	/**
// 	 * 准备开始处理数据。需要做一些数据初始化操作。
// 	 */
// 	public void prepare(TopologyContext context);
// 	
// 	
// 	/**
// 	 * 处理一个数据元素。
// 	 * 
// 	 * @param data
// 	 */
// 	public void execute(DataCell data, OutputCollector output);
// 
// 	/**
// 	 * 系统推出时调用。
// 	 */
// 	public void shutdown();
// 
// }
// 
// Path: src/main/java/com/taodian/monitor/storm/DataSpout.java
// public interface DataSpout {
// 	
// 	public void prepare(TopologyContext context);
// 	public DataCell nextDataCell();
// 	
// 	public boolean isClosed();
// }
// 
// Path: src/main/java/com/taodian/monitor/storm/Topology.java
// public interface Topology {
// 	
// 	/**
// 	 * 设置一个输入数据源，
// 	 * 
// 	 * @param name 输入数据源的名字。
// 	 * @param spout， 数据源
// 	 */
// 	public void setSpout(String name, DataSpout spout, String output);
// 
// 	public void setSpout(String name, DataSpout spout);
// 
// 	
// 	/**
// 	 * 设置一个输入数据源，
// 	 * 
// 	 * @param name 数据来源
// 	 * @param bolt，
// 	 * @param output -- 输出的消息队列。
// 	 */
// 	public void setBolt(String from, DataBolt bolt, String output);
// 	
// 	/**
// 	 * 开始运行
// 	 */
// 	public void start();
// 	
// 	public List<DataQueue> getAllDataQueue();
// }


// Path: src/main/java/com/taodian/monitor/storm/impl/SimpleTopologyBuilder.java
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.taodian.monitor.storm.DataBolt;
import com.taodian.monitor.storm.DataSpout;
import com.taodian.monitor.storm.ItemFactory;
import com.taodian.monitor.storm.Topology;
import com.taodian.monitor.storm.TopologyBuilder;

package com.taodian.monitor.storm.impl;




/**
 * 根据文本输入流配置拓扑。
 * @author deonwu
 */
public class SimpleTopologyBuilder implements TopologyBuilder{
	private Log log = LogFactory.getLog("storm.builder");  
	
	private BufferedReader reader = null;
	public SimpleTopologyBuilder(InputStream ins){
		try {
			reader = new BufferedReader(new InputStreamReader(ins, "utf8"));
		} catch (UnsupportedEncodingException e) {
			log.error(e, e);
		}
	}

	@Override
	public void buildToplogy(Topology topology, ItemFactory factory){
		try{
			log.info("-------Read topology configuration-------");
			for(String r = reader.readLine(); r != null; r = reader.readLine()){
				r = r.trim();
				if(r.startsWith("#") || r.length() == 0) continue;
				String[] tmp = r.split("=", 2);
				log.info(r);
				if(tmp.length == 2){
					String[] x = tmp[1].split(",", 2);
					Object obj = factory.create(x[0].trim());
					String p = x.length == 2 ? x[1] : "";
					if(obj instanceof DataSpout){
						topology.setSpout(tmp[0].trim(), (DataSpout)obj, p);