// 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 acromusashi.stream.bolt.hdfs.HdfsStoreBolt;
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;

/**
* Copyright (c) Acroquest Technology Co, Ltd. All Rights Reserved.
* Please read the associated COPYRIGHTS file for more details.
*
* THE SOFTWARE IS PROVIDED BY Acroquest Technolog Co., Ltd.,
* WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
* BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDER BE LIABLE FOR ANY
* CLAIM, DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THIS SOFTWARE OR ITS DERIVATIVES.
*/
package acromusashi.stream.example.topology;


/**
 * HDFS DataStore用のTopologyを起動する。
 * <br/>
 * Topologyの動作フローは下記の通り。<br/>
 * <ol>
 * <li>PeriodicalMessageGenSpoutにてメッセージを生成する</li>
 * <li>HdfsStoreBoltにて設定項目[hdfsstorebolt.～]に設定したHDFSに対してデータを投入する</li>
 * </ol>
 * 
 * <ul>
 * <li>MessageGenSpout.Parallelism : PeriodicalMessageGenSpoutの並列度(デフォルト値:1)</li>
 * <li>HdfsStoreBolt.Parallelism : HdfsStoreBoltの並列度(デフォルト値:1)</li>
 * <li>hdfsstorebolt.outputuri : HdfsStoreBoltにおける出力先URI(デフォルト値:無)</li>
 * <li>hdfsstorebolt.filenameheader : HDFSに出力するファイル名ヘッダ(デフォルト値:無)</li>
 * <li>hdfsstorebolt.interval : HDFSに出力する際のファイル切替インターバル(単位:分)(デフォルト値:10)</li>
 * <li>hdfsstorebolt.executepreprocess : 起動時HDFS上の.tmpファイルをリネームするか。Booleanオブジェクトを設定すること(デフォルト値:true)</li>
 * </ul>
 * @author otoda
 */
public class HdfsStoreTopology extends BaseTopology
{
    /**
     * コンストラクタ
     * 
     * @param topologyName Topology名称
     * @param config Storm設定オブジェクト
     */
    public HdfsStoreTopology(String topologyName, Config config)
    {
        super(topologyName, config);
    }

    /**
     * プログラムエントリポイント<br/>
     * <ul>
     * <li>起動引数:arg[0] 設定値を記述したyamlファイルパス</li>
     * <li>起動引数:arg[1] Stormの起動モード(true:LocalMode、false:DistributeMode)</li>
     * </ul>
     * @param args 起動引数
     * @throws Exception 初期化例外発生時
     */

// 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/HdfsStoreTopology.java
import acromusashi.stream.bolt.hdfs.HdfsStoreBolt;
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;

/**
* Copyright (c) Acroquest Technology Co, Ltd. All Rights Reserved.
* Please read the associated COPYRIGHTS file for more details.
*
* THE SOFTWARE IS PROVIDED BY Acroquest Technolog Co., Ltd.,
* WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
* BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDER BE LIABLE FOR ANY
* CLAIM, DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THIS SOFTWARE OR ITS DERIVATIVES.
*/
package acromusashi.stream.example.topology;


/**
 * HDFS DataStore用のTopologyを起動する。
 * <br/>
 * Topologyの動作フローは下記の通り。<br/>
 * <ol>
 * <li>PeriodicalMessageGenSpoutにてメッセージを生成する</li>
 * <li>HdfsStoreBoltにて設定項目[hdfsstorebolt.～]に設定したHDFSに対してデータを投入する</li>
 * </ol>
 * 
 * <ul>
 * <li>MessageGenSpout.Parallelism : PeriodicalMessageGenSpoutの並列度(デフォルト値:1)</li>
 * <li>HdfsStoreBolt.Parallelism : HdfsStoreBoltの並列度(デフォルト値:1)</li>
 * <li>hdfsstorebolt.outputuri : HdfsStoreBoltにおける出力先URI(デフォルト値:無)</li>
 * <li>hdfsstorebolt.filenameheader : HDFSに出力するファイル名ヘッダ(デフォルト値:無)</li>
 * <li>hdfsstorebolt.interval : HDFSに出力する際のファイル切替インターバル(単位:分)(デフォルト値:10)</li>
 * <li>hdfsstorebolt.executepreprocess : 起動時HDFS上の.tmpファイルをリネームするか。Booleanオブジェクトを設定すること(デフォルト値:true)</li>
 * </ul>
 * @author otoda
 */
public class HdfsStoreTopology extends BaseTopology
{
    /**
     * コンストラクタ
     * 
     * @param topologyName Topology名称
     * @param config Storm設定オブジェクト
     */
    public HdfsStoreTopology(String topologyName, Config config)
    {
        super(topologyName, config);
    }

    /**
     * プログラムエントリポイント<br/>
     * <ul>
     * <li>起動引数:arg[0] 設定値を記述したyamlファイルパス</li>
     * <li>起動引数:arg[1] Stormの起動モード(true:LocalMode、false:DistributeMode)</li>
     * </ul>
     * @param args 起動引数
     * @throws Exception 初期化例外発生時
     */

    public static void main(String[] args) throws Exception
