// Path: src/main/java/storm/benchmark/api/IApplication.java
// public interface IApplication {
//   public StormTopology getTopology(Config config);
// }
// 
// Path: src/main/java/storm/benchmark/api/IBenchmark.java
// public interface IBenchmark extends IApplication {
//   public IMetricsCollector getMetricsCollector(Config config, StormTopology topology);
// }
// 
// Path: src/main/java/storm/benchmark/api/IProducer.java
// public interface IProducer extends IApplication {
// }
// 
// Path: src/main/java/storm/benchmark/metrics/MetricsCollectorConfig.java
// public class MetricsCollectorConfig {
//   private static final Logger LOG = Logger.getLogger(MetricsCollectorConfig.class);
// 
//   public static final String CONF_FILE_FORMAT = "%s/%s_metrics_%d.yaml";
//   public static final String DATA_FILE_FORMAT = "%s/%s_metrics_%d.csv";
// 
//   public static final String METRICS_ENABLED = "metrics.enabled";
//   public static final String METRICS_POLL_INTERVAL = "metrics.poll";
//   public static final String METRICS_TOTAL_TIME = "metrics.time";
//   public static final String METRICS_PATH = "metrics.path";
// 
//   public static final int DEFAULT_POLL_INTERVAL = 30 * 1000; // 30 secs
//   public static final int DEFAULT_TOTAL_TIME = 5 * 60 * 1000; // 5 mins
//   public static final String DEFAULT_PATH = "/root/";
// 
//   // storm configuration
//   public final Config stormConfig;
//   // storm benchmarks name
//   public final String name;
//   // How often should metrics be collected
//   public final int pollInterval;
//   // How long should the benchmark run for
//   public final int totalTime;
//   // metrics file path
//   public final String path;
// 
//   public MetricsCollectorConfig(Config stormConfig) {
//     this.stormConfig  = stormConfig;
//     name = (String) Utils.get(
//             stormConfig, Config.TOPOLOGY_NAME, StormBenchmark.DEFAULT_TOPOLOGY_NAME);
//     pollInterval = BenchmarkUtils.getInt(
//             stormConfig, METRICS_POLL_INTERVAL, DEFAULT_POLL_INTERVAL);
//     totalTime = BenchmarkUtils.getInt(
//             stormConfig, METRICS_TOTAL_TIME, DEFAULT_TOTAL_TIME);
//     path = (String) Utils.get(stormConfig, METRICS_PATH, DEFAULT_PATH);
//   }
// 
// 
//   public void writeStormConfig(PrintWriter writer) {
//     LOG.info("writing out storm config into .yaml file");
//     if (writer != null) {
//       Map sorted = new TreeMap();
//       sorted.putAll(stormConfig);
//       for (Object key : sorted.keySet()) {
//         writer.println(key + ": " + stormConfig.get(key));
//       }
//       writer.flush();
//     }
//   }
// }


import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.utils.Utils;
import org.apache.log4j.Logger;
import storm.benchmark.api.IApplication;
import storm.benchmark.api.IBenchmark;
import storm.benchmark.api.IProducer;
import storm.benchmark.metrics.IMetricsCollector;
import storm.benchmark.metrics.MetricsCollectorConfig;
    if (name.startsWith("storm.benchmark.benchmarks")) {
      LOG.info("running benchmark " + name);
      runBenchmark((IBenchmark) getApplicationFromName(name));
    } else if (name.startsWith("storm.benchmark.tools.producer")) {
      LOG.info("running producer " + name);
      runProducer((IProducer) getApplicationFromName(name));
    } else {
      throw new RuntimeException(name + " is neither benchmark nor producer");
    }
  }

  public static void runBenchmark(IBenchmark benchmark)
          throws AlreadyAliveException, InvalidTopologyException,
          ClassNotFoundException, IllegalAccessException, InstantiationException {
    runApplication(benchmark);
    if (isMetricsEnabled()) {
      IMetricsCollector collector = benchmark.getMetricsCollector(config, topology);
      collector.run();
    }
  }

  public static void runProducer(IProducer producer)
          throws AlreadyAliveException, InvalidTopologyException,
          ClassNotFoundException, IllegalAccessException, InstantiationException {
    runApplication(producer);
  }


  public static IApplication getApplicationFromName(String name)
          throws ClassNotFoundException, IllegalAccessException, InstantiationException {
// Path: src/main/java/storm/benchmark/api/IApplication.java
// public interface IApplication {
//   public StormTopology getTopology(Config config);
// }
// 
// Path: src/main/java/storm/benchmark/api/IBenchmark.java
// public interface IBenchmark extends IApplication {
//   public IMetricsCollector getMetricsCollector(Config config, StormTopology topology);
// }
// 
// Path: src/main/java/storm/benchmark/api/IProducer.java
// public interface IProducer extends IApplication {
// }
// 
// Path: src/main/java/storm/benchmark/metrics/MetricsCollectorConfig.java
// public class MetricsCollectorConfig {
//   private static final Logger LOG = Logger.getLogger(MetricsCollectorConfig.class);
// 
//   public static final String CONF_FILE_FORMAT = "%s/%s_metrics_%d.yaml";
//   public static final String DATA_FILE_FORMAT = "%s/%s_metrics_%d.csv";
// 
//   public static final String METRICS_ENABLED = "metrics.enabled";
//   public static final String METRICS_POLL_INTERVAL = "metrics.poll";
//   public static final String METRICS_TOTAL_TIME = "metrics.time";
//   public static final String METRICS_PATH = "metrics.path";
// 
//   public static final int DEFAULT_POLL_INTERVAL = 30 * 1000; // 30 secs
//   public static final int DEFAULT_TOTAL_TIME = 5 * 60 * 1000; // 5 mins
//   public static final String DEFAULT_PATH = "/root/";
// 
//   // storm configuration
//   public final Config stormConfig;
//   // storm benchmarks name
//   public final String name;
//   // How often should metrics be collected
//   public final int pollInterval;
//   // How long should the benchmark run for
//   public final int totalTime;
//   // metrics file path
//   public final String path;
// 
//   public MetricsCollectorConfig(Config stormConfig) {
//     this.stormConfig  = stormConfig;
//     name = (String) Utils.get(
//             stormConfig, Config.TOPOLOGY_NAME, StormBenchmark.DEFAULT_TOPOLOGY_NAME);
//     pollInterval = BenchmarkUtils.getInt(
//             stormConfig, METRICS_POLL_INTERVAL, DEFAULT_POLL_INTERVAL);
//     totalTime = BenchmarkUtils.getInt(
//             stormConfig, METRICS_TOTAL_TIME, DEFAULT_TOTAL_TIME);
//     path = (String) Utils.get(stormConfig, METRICS_PATH, DEFAULT_PATH);
//   }
// 
// 
//   public void writeStormConfig(PrintWriter writer) {
//     LOG.info("writing out storm config into .yaml file");
//     if (writer != null) {
//       Map sorted = new TreeMap();
//       sorted.putAll(stormConfig);
//       for (Object key : sorted.keySet()) {
//         writer.println(key + ": " + stormConfig.get(key));
//       }
//       writer.flush();
//     }
//   }
// }


// Path: src/main/java/storm/benchmark/tools/Runner.java
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.utils.Utils;
import org.apache.log4j.Logger;
import storm.benchmark.api.IApplication;
import storm.benchmark.api.IBenchmark;
import storm.benchmark.api.IProducer;
import storm.benchmark.metrics.IMetricsCollector;
import storm.benchmark.metrics.MetricsCollectorConfig;

    if (name.startsWith("storm.benchmark.benchmarks")) {
      LOG.info("running benchmark " + name);
      runBenchmark((IBenchmark) getApplicationFromName(name));
    } else if (name.startsWith("storm.benchmark.tools.producer")) {
      LOG.info("running producer " + name);
      runProducer((IProducer) getApplicationFromName(name));
    } else {
      throw new RuntimeException(name + " is neither benchmark nor producer");
    }
  }

  public static void runBenchmark(IBenchmark benchmark)
          throws AlreadyAliveException, InvalidTopologyException,
          ClassNotFoundException, IllegalAccessException, InstantiationException {
    runApplication(benchmark);
    if (isMetricsEnabled()) {
      IMetricsCollector collector = benchmark.getMetricsCollector(config, topology);
      collector.run();
    }
  }

  public static void runProducer(IProducer producer)
          throws AlreadyAliveException, InvalidTopologyException,
          ClassNotFoundException, IllegalAccessException, InstantiationException {
    runApplication(producer);
  }


  public static IApplication getApplicationFromName(String name)
          throws ClassNotFoundException, IllegalAccessException, InstantiationException {
    return (IApplication) Class.forName(name).newInstance();