// Path: src/main/java/acromusashi/stream/example/bolt/RedisBolt.java
// public class RedisBolt extends MessageBolt
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = 7742873990972648063L;
// 
//     /** redishost */
//     private String              redisHost;
// 
//     /** Redisにアクセスするプールオブジェクト */
//     private transient JedisPool jedisPool;
// 
//     /** Redisにアクセスするクライアントオブジェクト */
//     private transient Jedis     jedisClient;
// 
//     /**
//      * コンストラクタ
//      * 
//      * @param redisHost Redis投入先ホスト
//      */
//     public RedisBolt(String redisHost)
//     {
//         this.redisHost = redisHost;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
//     {
//         super.prepare(stormConf, context, collector);
//         this.jedisPool = new JedisPool(this.redisHost);
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void onMessage(Message message) throws Exception
//     {
//         if (this.jedisClient == null)
//         {
//             this.jedisClient = this.jedisPool.getResource();
//         }
// 
//         this.jedisClient.set(message.getHeader().getMessageId(), message.getBody().toString());
//     }
// }


import java.util.List;
import acromusashi.stream.bolt.MessageConvertBolt;
import acromusashi.stream.component.kestrel.spout.KestrelSpout;
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.bolt.RedisBolt;
import acromusashi.stream.topology.BaseTopology;
import backtype.storm.Config;
import backtype.storm.scheme.StringScheme;
import backtype.storm.spout.SchemeAsMultiScheme;
        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);
        String redisHost = StormConfigUtil.getStringValue(getConfig(), "RedisBolt.RedisHost",
                "localhost");
        int redisBoltPara = StormConfigUtil.getIntValue(getConfig(), "RedisBolt.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 SnmpConverter());
        getBuilder().setBolt("ConvertBolt", convertBolt, msgConvertPara).localOrShuffleGrouping(
                "KestrelSpout");

        // Add Bolt(MessageConvertBolt -> RedisBolt)
// Path: src/main/java/acromusashi/stream/example/bolt/RedisBolt.java
// public class RedisBolt extends MessageBolt
// {
//     /** serialVersionUID */
//     private static final long   serialVersionUID = 7742873990972648063L;
// 
//     /** redishost */
//     private String              redisHost;
// 
//     /** Redisにアクセスするプールオブジェクト */
//     private transient JedisPool jedisPool;
// 
//     /** Redisにアクセスするクライアントオブジェクト */
//     private transient Jedis     jedisClient;
// 
//     /**
//      * コンストラクタ
//      * 
//      * @param redisHost Redis投入先ホスト
//      */
//     public RedisBolt(String redisHost)
//     {
//         this.redisHost = redisHost;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
//     {
//         super.prepare(stormConf, context, collector);
//         this.jedisPool = new JedisPool(this.redisHost);
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void onMessage(Message message) throws Exception
//     {
//         if (this.jedisClient == null)
//         {
//             this.jedisClient = this.jedisPool.getResource();
//         }
// 
//         this.jedisClient.set(message.getHeader().getMessageId(), message.getBody().toString());
//     }
// }


// Path: src/main/java/acromusashi/stream/example/topology/Snmp2RedisTopology.java
import java.util.List;
import acromusashi.stream.bolt.MessageConvertBolt;
import acromusashi.stream.component.kestrel.spout.KestrelSpout;
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.bolt.RedisBolt;
import acromusashi.stream.topology.BaseTopology;
import backtype.storm.Config;
import backtype.storm.scheme.StringScheme;
import backtype.storm.spout.SchemeAsMultiScheme;

        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);
        String redisHost = StormConfigUtil.getStringValue(getConfig(), "RedisBolt.RedisHost",
                "localhost");
        int redisBoltPara = StormConfigUtil.getIntValue(getConfig(), "RedisBolt.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 SnmpConverter());
        getBuilder().setBolt("ConvertBolt", convertBolt, msgConvertPara).localOrShuffleGrouping(
                "KestrelSpout");

        // Add Bolt(MessageConvertBolt -> RedisBolt)
        RedisBolt bolt = new RedisBolt(redisHost);