/**
* 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.topology.state;

import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;

import acromusashi.stream.client.NimbusClientFactory;
import acromusashi.stream.exception.ConnectFailException;
import backtype.storm.generated.ClusterSummary;
import backtype.storm.generated.Nimbus;
import backtype.storm.generated.NotAliveException;
import backtype.storm.generated.TopologyInfo;
import backtype.storm.utils.NimbusClient;

/**
 * Nimbusに問合せ、該当Topologyの実行状態を取得するクライアントクラス<br>
 * 該当Topologyのメッセージ処理数を定期的に確認し、指定した判定実行回数分変化がなかった場合に該当Topologyを「停止中」として扱う。<br>
 * メッセージ処理数に変化があった場合はメッセージ処理中として扱い、「処理中」として扱う。
 *
 * @author kimura
 */
public class TopologyExecutionGetter
{
    /**
     * 外部からのインスタンス化を防止するデフォルトコンストラクタ。
     */
    private TopologyExecutionGetter()
    {
        // Do nothing.
    }

    /**
     * 該当Topologyの実行状態を取得する。
     *
     * @param topologyName トポロジ名
     * @param nimbusHost NimbusHost
     * @param nimbusPort NimbusPort
     * @param interval 処理中判定実行インターバル（ミリ秒単位）
     * @param number 処理中判定実行回数
     * @throws ConnectFailException 接続失敗時
     * @return Topology実行状態
     */
    public TopologyExecutionStatus getExecution(String topologyName, String nimbusHost,
            int nimbusPort, int interval, int number) throws ConnectFailException
    {

        NimbusClientFactory factory = new NimbusClientFactory();
        // RuntimeExceptionが発生
        NimbusClient nimbusClient = factory.createClient(nimbusHost, nimbusPort);
        Nimbus.Client client = nimbusClient.getClient();
        ClusterSummary clusterSummary = null;
        try
        {
            clusterSummary = client.getClusterInfo();
        }
        catch (org.apache.thrift7.TException ex)
        {
            throw new ConnectFailException(ex);
        }

        String topologyId = StormSummaryExtractor.extractStormTopologyId(clusterSummary,
                topologyName);

        if (topologyId == null)
        {
            return TopologyExecutionStatus.NOT_ALIVED;
        }

        TopologyExecutionCount compareCount = null;

        for (int getCount = 0; getCount < number; getCount++)
        {
            TopologyInfo topologyInfo = null;

            try
            {
                topologyInfo = client.getTopologyInfo(topologyId);
            }
            catch (NotAliveException ex)
            {
                return TopologyExecutionStatus.NOT_ALIVED;
            }
            catch (org.apache.thrift7.TException ex)
            {
                throw new ConnectFailException(ex);
            }

            TopologyExecutionCount nowCount = StormSummaryExtractor.convertToTotalCount(topologyInfo);
            // 初回の情報取得の場合、初回の取得結果を比較対象として設定
            if (compareCount == null)
            {
                compareCount = nowCount;
            }
            else
            {
                // 文字列表現が異なる場合、処理が継続しているため、ログ出力して「処理中」と返す。
                if (StringUtils.equals(compareCount.toString(), nowCount.toString()) == false)
                {
                    return TopologyExecutionStatus.EXECUTING;
                }
            }

            try
            {
                TimeUnit.MILLISECONDS.sleep(interval);
            }
            catch (InterruptedException ex)
            {
                continue;
            }
        }

        // 指定回数変化がなかった場合、停止として扱う。
        return TopologyExecutionStatus.STOP;
    }
}

@@ -1288,37 +1288,30 @@
   * 
-%E5%A4%96%E9%83%A8%E3%81%8B%E3%82%89%E3%81%AE%E3%82%A4%E3%83%B3%E3%82%B9%E3%82%BF%E3%83%B3%E3%82%B9%E5%8C%96%E3%82%92%E9%98%B2%E6%AD%A2%E3%81%99%E3%82%8B%E3%83%87%E3%83%95%E3%82%A9%E3%83%AB%E3%83%88%E3%82%B3%E3%83%B3%E3%82%B9%E3%83%88%E3%83%A9%E3%82%AF%E3%82%BF
+%E3%83%91%E3%83%A9%E3%83%A1%E3%83%BC%E3%82%BF%E3%82%92%E6%8C%87%E5%AE%9A%E3%81%9B%E3%81%9A%E3%81%AB%E3%82%A4%E3%83%B3%E3%82%B9%E3%82%BF%E3%83%B3%E3%82%B9%E3%82%92%E7%94%9F%E6%88%90%E3%81%99%E3%82%8B
 %E3%80%82%0A  
@@ -1325,14 +1325,13 @@
    p
-rivate
+ublic
  Top

Modify access level