// Path: src/main/java/acromusashi/stream/example/bolt/ConsolePrintBolt.java
// public class ConsolePrintBolt extends AmBaseBolt
// {
//     /** serialVersionUID */
//     private static final long serialVersionUID = 5100460578090478268L;
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public ConsolePrintBolt()
//     {}
// 
//     /**
//      * {@inheritDoc}
//      */
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void onPrepare(Map config, TopologyContext context)
//     {
//         // TODO Auto-generated method stub
// 
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void onExecute(StreamMessage message)
//     {
//         System.out.println(message.toString());
//     }
// }


import java.util.List;
import acromusashi.stream.bolt.MessageConvertBolt;
import acromusashi.stream.component.kestrel.spout.KestrelSpout;
import acromusashi.stream.component.twitter.converter.TwitterJsonConverter;
import acromusashi.stream.config.StormConfigGenerator;
import acromusashi.stream.config.StormConfigUtil;
import acromusashi.stream.entity.Message;
import acromusashi.stream.example.bolt.ConsolePrintBolt;
import acromusashi.stream.topology.BaseTopology;
import backtype.storm.Config;
import backtype.storm.scheme.StringScheme;
import backtype.storm.spout.SchemeAsMultiScheme;

        topology.buildTopology();
        topology.submitTopology(isLocal);
    }

    @Override
    public void buildTopology() throws Exception
    {
        // Get setting from StormConfig Object
        List<String> kestrelHosts = StormConfigUtil.getStringListValue(getConfig(), "Kestrel.Hosts");
        String kestrelQueueName = StormConfigUtil.getStringValue(getConfig(), "Kestrel.QueueName",
                "MessageQueue");
        int kestrelSpoutPara = StormConfigUtil.getIntValue(getConfig(), "KestrelSpout.Parallelism",
                1);
        int msgConvertPara = StormConfigUtil.getIntValue(getConfig(), "ConvertBolt.Parallelism", 1);
        int consoleBoltPara = StormConfigUtil.getIntValue(getConfig(),
                "ConsolePrintBolt.Parallelism", 1);

        // Topology Setting
        // Add Spout(KestrelSpout)
        KestrelSpout kestrelSpout = new KestrelSpout(kestrelHosts, kestrelQueueName,
                new SchemeAsMultiScheme(new StringScheme()));
        getBuilder().setSpout("KestrelSpout", kestrelSpout, kestrelSpoutPara);

        // Add Bolt(KestrelSpout -> MessageConvertBolt)
        MessageConvertBolt convertBolt = new MessageConvertBolt();
        convertBolt.setConverter(new TwitterJsonConverter());
        getBuilder().setBolt("ConvertBolt", convertBolt, msgConvertPara).localOrShuffleGrouping(
                "KestrelSpout");

        // Add Bolt(MessageConvertBolt -> ConsolePrintBolt)

// Path: src/main/java/acromusashi/stream/example/bolt/ConsolePrintBolt.java
// public class ConsolePrintBolt extends AmBaseBolt
// {
//     /** serialVersionUID */
//     private static final long serialVersionUID = 5100460578090478268L;
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public ConsolePrintBolt()
//     {}
// 
//     /**
//      * {@inheritDoc}
//      */
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void onPrepare(Map config, TopologyContext context)
//     {
//         // TODO Auto-generated method stub
// 
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void onExecute(StreamMessage message)
//     {
//         System.out.println(message.toString());
//     }
// }


// Path: src/main/java/acromusashi/stream/example/topology/TwitterJsonConsolePrintTopology.java
import java.util.List;
import acromusashi.stream.bolt.MessageConvertBolt;
import acromusashi.stream.component.kestrel.spout.KestrelSpout;
import acromusashi.stream.component.twitter.converter.TwitterJsonConverter;
import acromusashi.stream.config.StormConfigGenerator;
import acromusashi.stream.config.StormConfigUtil;
import acromusashi.stream.entity.Message;
import acromusashi.stream.example.bolt.ConsolePrintBolt;
import acromusashi.stream.topology.BaseTopology;
import backtype.storm.Config;
import backtype.storm.scheme.StringScheme;
import backtype.storm.spout.SchemeAsMultiScheme;

        topology.buildTopology();
        topology.submitTopology(isLocal);
    }

    @Override
    public void buildTopology() throws Exception
    {
        // Get setting from StormConfig Object
        List<String> kestrelHosts = StormConfigUtil.getStringListValue(getConfig(), "Kestrel.Hosts");
        String kestrelQueueName = StormConfigUtil.getStringValue(getConfig(), "Kestrel.QueueName",
                "MessageQueue");
        int kestrelSpoutPara = StormConfigUtil.getIntValue(getConfig(), "KestrelSpout.Parallelism",
                1);
        int msgConvertPara = StormConfigUtil.getIntValue(getConfig(), "ConvertBolt.Parallelism", 1);
        int consoleBoltPara = StormConfigUtil.getIntValue(getConfig(),
                "ConsolePrintBolt.Parallelism", 1);

        // Topology Setting
        // Add Spout(KestrelSpout)
        KestrelSpout kestrelSpout = new KestrelSpout(kestrelHosts, kestrelQueueName,
                new SchemeAsMultiScheme(new StringScheme()));
        getBuilder().setSpout("KestrelSpout", kestrelSpout, kestrelSpoutPara);

        // Add Bolt(KestrelSpout -> MessageConvertBolt)
        MessageConvertBolt convertBolt = new MessageConvertBolt();
        convertBolt.setConverter(new TwitterJsonConverter());
        getBuilder().setBolt("ConvertBolt", convertBolt, msgConvertPara).localOrShuffleGrouping(
                "KestrelSpout");

        // Add Bolt(MessageConvertBolt -> ConsolePrintBolt)

        ConsolePrintBolt bolt = new ConsolePrintBolt();
