// Path: src/main/java/acromusashi/stream/example/spout/PeriodicalMessageGenSpout.java
// public class PeriodicalMessageGenSpout extends BaseConfigurationSpout
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = -237111294339742815L;
// 
//     /** logger */
//     private static final Logger logger           = LoggerFactory.getLogger(PeriodicalMessageGenSpout.class);
// 
//     /** 送信カウンタ */
//     private int                 counter          = 0;
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public PeriodicalMessageGenSpout()
//     {}
// 
//     /**
//      * 一定間隔ごとにデータをBoltに送信する
//      */
//     @Override
//     public void nextTuple()
//     {
//         this.counter++;
//         Header header = new Header();
//         header.setMessageId(UUID.randomUUID().toString());
//         header.setTimestamp(System.currentTimeMillis());
//         header.setSource("192.168.0.1");
//         header.setType("message");
//         Message message = new Message();
//         message.setHeader(header);
// 
//         List<Object> list = new ArrayList<Object>();
//         list.add("Counter:");
//         list.add(this.counter);
//         message.setBody(list);
// 
//         getCollector().emit(new Values(message));
// 
//         try
//         {
//             TimeUnit.SECONDS.sleep(1);
//         }
//         catch (InterruptedException iex)
//         {
//             if (logger.isDebugEnabled() == true)
//             {
//                 logger.debug("Occur interrupt. Ignore interrupt.", iex);
//             }
//         }
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void declareOutputFields(OutputFieldsDeclarer declarer)
//     {
//         declarer.declare(new Fields("message"));
//     }
// }


import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import acromusashi.stream.bolt.hbase.CamelHbaseStoreBolt;
import acromusashi.stream.bolt.hbase.CellDefine;
import acromusashi.stream.component.snmp.converter.SnmpConverter;
import acromusashi.stream.config.StormConfigGenerator;
import acromusashi.stream.config.StormConfigUtil;
import acromusashi.stream.entity.Message;
import acromusashi.stream.example.spout.PeriodicalMessageGenSpout;
import acromusashi.stream.topology.BaseTopology;
import backtype.storm.Config;

            return;
        }

        // 起動引数として使用したパスからStorm設定オブジェクトを生成
        Config conf = StormConfigGenerator.loadStormConfig(args[0]);

        // プログラム引数から設定値を取得(ローカル環境or分散環境)
        boolean isLocal = Boolean.valueOf(args[1]);

        // Topologyを起動する
        BaseTopology topology = new HbaseStoreTopology("HbaseStoreTopology", conf);
        topology.buildTopology();
        topology.submitTopology(isLocal);
    }

    @Override
    public void buildTopology() throws Exception
    {
        // Get setting from StormConfig Object
        int msgGenSpoutPara = StormConfigUtil.getIntValue(getConfig(),
                "MessageGenSpout.Parallelism", 1);
        int hbaseBoltPara = StormConfigUtil.getIntValue(getConfig(), "CamelHBaseBolt.Parallelism",
                1);
        String contextUri = StormConfigUtil.getStringValue(getConfig(), "CamelContext.Path",
                "file:/opt/storm/conf/camel-context-example-hbase.xml");
        List<String> cellDefineList = StormConfigUtil.getStringListValue(getConfig(),
                "HBaseSchema.Define");

        // Topology Setting
        // Add Spout(PeriodicalMessageGenSpout)

// Path: src/main/java/acromusashi/stream/example/spout/PeriodicalMessageGenSpout.java
// public class PeriodicalMessageGenSpout extends BaseConfigurationSpout
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = -237111294339742815L;
// 
//     /** logger */
//     private static final Logger logger           = LoggerFactory.getLogger(PeriodicalMessageGenSpout.class);
// 
//     /** 送信カウンタ */
//     private int                 counter          = 0;
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public PeriodicalMessageGenSpout()
//     {}
// 
//     /**
//      * 一定間隔ごとにデータをBoltに送信する
//      */
//     @Override
//     public void nextTuple()
//     {
//         this.counter++;
//         Header header = new Header();
//         header.setMessageId(UUID.randomUUID().toString());
//         header.setTimestamp(System.currentTimeMillis());
//         header.setSource("192.168.0.1");
//         header.setType("message");
//         Message message = new Message();
//         message.setHeader(header);
// 
//         List<Object> list = new ArrayList<Object>();
//         list.add("Counter:");
//         list.add(this.counter);
//         message.setBody(list);
// 
//         getCollector().emit(new Values(message));
// 
//         try
//         {
//             TimeUnit.SECONDS.sleep(1);
//         }
//         catch (InterruptedException iex)
//         {
//             if (logger.isDebugEnabled() == true)
//             {
//                 logger.debug("Occur interrupt. Ignore interrupt.", iex);
//             }
//         }
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void declareOutputFields(OutputFieldsDeclarer declarer)
//     {
//         declarer.declare(new Fields("message"));
//     }
// }


// Path: src/main/java/acromusashi/stream/example/topology/HbaseStoreTopology.java
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import acromusashi.stream.bolt.hbase.CamelHbaseStoreBolt;
import acromusashi.stream.bolt.hbase.CellDefine;
import acromusashi.stream.component.snmp.converter.SnmpConverter;
import acromusashi.stream.config.StormConfigGenerator;
import acromusashi.stream.config.StormConfigUtil;
import acromusashi.stream.entity.Message;
import acromusashi.stream.example.spout.PeriodicalMessageGenSpout;
import acromusashi.stream.topology.BaseTopology;
import backtype.storm.Config;

            return;
        }

        // 起動引数として使用したパスからStorm設定オブジェクトを生成
        Config conf = StormConfigGenerator.loadStormConfig(args[0]);

        // プログラム引数から設定値を取得(ローカル環境or分散環境)
        boolean isLocal = Boolean.valueOf(args[1]);

        // Topologyを起動する
        BaseTopology topology = new HbaseStoreTopology("HbaseStoreTopology", conf);
        topology.buildTopology();
        topology.submitTopology(isLocal);
    }

    @Override
    public void buildTopology() throws Exception
    {
        // Get setting from StormConfig Object
        int msgGenSpoutPara = StormConfigUtil.getIntValue(getConfig(),
                "MessageGenSpout.Parallelism", 1);
        int hbaseBoltPara = StormConfigUtil.getIntValue(getConfig(), "CamelHBaseBolt.Parallelism",
                1);
        String contextUri = StormConfigUtil.getStringValue(getConfig(), "CamelContext.Path",
                "file:/opt/storm/conf/camel-context-example-hbase.xml");
        List<String> cellDefineList = StormConfigUtil.getStringListValue(getConfig(),
                "HBaseSchema.Define");

        // Topology Setting
        // Add Spout(PeriodicalMessageGenSpout)

        PeriodicalMessageGenSpout messageGenSpout = new PeriodicalMessageGenSpout();
