// Path: src/main/java/storm/benchmark/benchmarks/common/StormBenchmark.java
// public abstract class StormBenchmark implements IBenchmark {
// 
//   private static final Logger LOG = Logger.getLogger(StormBenchmark.class);
//   public static final String DEFAULT_TOPOLOGY_NAME = "benchmark";
// 
//   @Override
//   public IMetricsCollector getMetricsCollector(Config config, StormTopology topology) {
// 
//     Set<MetricsItem> items = Sets.newHashSet(
//             MetricsItem.SUPERVISOR_STATS,
//             MetricsItem.TOPOLOGY_STATS,
//             MetricsItem.THROUGHPUT,
//             MetricsItem.SPOUT_THROUGHPUT,
//             MetricsItem.SPOUT_LATENCY
//             );
//     return new BasicMetricsCollector(config, topology, items);
//   }
// 
// 
// }
// 
// Path: src/main/java/storm/benchmark/util/BenchmarkUtils.java
// public final class BenchmarkUtils {
//   private static final Logger LOG = Logger.getLogger(BenchmarkUtils.class);
// 
//   private BenchmarkUtils() {
//   }
// 
//   public static double max(Iterable<Double> iterable) {
//     Iterator<Double> iterator = iterable.iterator();
//     double max = Double.MIN_VALUE;
//     while (iterator.hasNext()) {
//       double d = iterator.next();
//       if (d > max) {
//         max = d;
//       }
//     }
//     return max;
//   }
// 
//   public static double avg(Iterable<Double> iterable) {
//     Iterator<Double> iterator = iterable.iterator();
//     double total = 0.0;
//     int num = 0;
//     while (iterator.hasNext()) {
//       total += iterator.next();
//       num++;
//     }
//     return total / num;
//   }
// 
//   public static void putIfAbsent(Map map, Object key, Object val) {
//     if (!map.containsKey(key)) {
//       map.put(key, val);
//     }
//   }
// 
//   public static int getInt(Map map, Object key, int def) {
//     return Utils.getInt(Utils.get(map, key, def));
//   }
// 
//   public static boolean ifAckEnabled(Config config) {
//     Object ackers = config.get(Config.TOPOLOGY_ACKER_EXECUTORS);
//     if (null == ackers) {
//       LOG.warn("acker executors are null");
//       return false;
//     }
//     return Utils.getInt(ackers) > 0;
//   }}
// 
// Path: src/main/java/storm/benchmark/util/TupleHelpers.java
// public final class TupleHelpers {
// 
//   private TupleHelpers() {
//   }
// 
//   public static boolean isTickTuple(Tuple tuple) {
//     return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
//         Constants.SYSTEM_TICK_STREAM_ID);
//   }
// 
// }


import storm.benchmark.benchmarks.common.StormBenchmark;
import storm.benchmark.lib.spout.RandomMessageSpout;
import storm.benchmark.util.BenchmarkUtils;
import storm.benchmark.util.TupleHelpers;
import java.io.Serializable;
import java.util.*;
import backtype.storm.Config;
import backtype.storm.generated.StormTopology;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import org.apache.log4j.Logger;

  public static class SortBolt extends BaseBasicBolt {

    public static final String EMIT_FREQ = "emit.frequency";
    public static final int DEFAULT_EMIT_FREQ = 60;  // 60s
    public static final String CHUNK_SIZE = "chunk.size";
    public static final int DEFAULT_CHUNK_SIZE = 100;
    public static final String FIELDS = "sorted_data";

    private final int emitFrequencyInSeconds;
    private final int chunkSize;
    private int index = 0;
    private MutableComparable[] data;


    public SortBolt(int emitFrequencyInSeconds, int chunkSize) {
      this.emitFrequencyInSeconds = emitFrequencyInSeconds;
      this.chunkSize = chunkSize;
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
      this.data = new MutableComparable[this.chunkSize];
      for (int i = 0; i < this.chunkSize; i++) {
        this.data[i] = new MutableComparable();
      }
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
// Path: src/main/java/storm/benchmark/benchmarks/common/StormBenchmark.java
// public abstract class StormBenchmark implements IBenchmark {
// 
//   private static final Logger LOG = Logger.getLogger(StormBenchmark.class);
//   public static final String DEFAULT_TOPOLOGY_NAME = "benchmark";
// 
//   @Override
//   public IMetricsCollector getMetricsCollector(Config config, StormTopology topology) {
// 
//     Set<MetricsItem> items = Sets.newHashSet(
//             MetricsItem.SUPERVISOR_STATS,
//             MetricsItem.TOPOLOGY_STATS,
//             MetricsItem.THROUGHPUT,
//             MetricsItem.SPOUT_THROUGHPUT,
//             MetricsItem.SPOUT_LATENCY
//             );
//     return new BasicMetricsCollector(config, topology, items);
//   }
// 
// 
// }
// 
// Path: src/main/java/storm/benchmark/util/BenchmarkUtils.java
// public final class BenchmarkUtils {
//   private static final Logger LOG = Logger.getLogger(BenchmarkUtils.class);
// 
//   private BenchmarkUtils() {
//   }
// 
//   public static double max(Iterable<Double> iterable) {
//     Iterator<Double> iterator = iterable.iterator();
//     double max = Double.MIN_VALUE;
//     while (iterator.hasNext()) {
//       double d = iterator.next();
//       if (d > max) {
//         max = d;
//       }
//     }
//     return max;
//   }
// 
//   public static double avg(Iterable<Double> iterable) {
//     Iterator<Double> iterator = iterable.iterator();
//     double total = 0.0;
//     int num = 0;
//     while (iterator.hasNext()) {
//       total += iterator.next();
//       num++;
//     }
//     return total / num;
//   }
// 
//   public static void putIfAbsent(Map map, Object key, Object val) {
//     if (!map.containsKey(key)) {
//       map.put(key, val);
//     }
//   }
// 
//   public static int getInt(Map map, Object key, int def) {
//     return Utils.getInt(Utils.get(map, key, def));
//   }
// 
//   public static boolean ifAckEnabled(Config config) {
//     Object ackers = config.get(Config.TOPOLOGY_ACKER_EXECUTORS);
//     if (null == ackers) {
//       LOG.warn("acker executors are null");
//       return false;
//     }
//     return Utils.getInt(ackers) > 0;
//   }}
// 
// Path: src/main/java/storm/benchmark/util/TupleHelpers.java
// public final class TupleHelpers {
// 
//   private TupleHelpers() {
//   }
// 
//   public static boolean isTickTuple(Tuple tuple) {
//     return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
//         Constants.SYSTEM_TICK_STREAM_ID);
//   }
// 
// }


// Path: src/main/java/storm/benchmark/benchmarks/RollingSort.java
import storm.benchmark.benchmarks.common.StormBenchmark;
import storm.benchmark.lib.spout.RandomMessageSpout;
import storm.benchmark.util.BenchmarkUtils;
import storm.benchmark.util.TupleHelpers;
import java.io.Serializable;
import java.util.*;
import backtype.storm.Config;
import backtype.storm.generated.StormTopology;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import org.apache.log4j.Logger;


  public static class SortBolt extends BaseBasicBolt {

    public static final String EMIT_FREQ = "emit.frequency";
    public static final int DEFAULT_EMIT_FREQ = 60;  // 60s
    public static final String CHUNK_SIZE = "chunk.size";
    public static final int DEFAULT_CHUNK_SIZE = 100;
    public static final String FIELDS = "sorted_data";

    private final int emitFrequencyInSeconds;
    private final int chunkSize;
    private int index = 0;
    private MutableComparable[] data;


    public SortBolt(int emitFrequencyInSeconds, int chunkSize) {
      this.emitFrequencyInSeconds = emitFrequencyInSeconds;
      this.chunkSize = chunkSize;
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
      this.data = new MutableComparable[this.chunkSize];
      for (int i = 0; i < this.chunkSize; i++) {
        this.data[i] = new MutableComparable();
      }
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
      if (TupleHelpers.isTickTuple(tuple)) {