class RollingCount extends StormBenchmark {

  @Override
  public StormTopology getTopology(Config config) {

    final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
    final int spBoltNum = BenchmarkUtils.getInt(config, SPLIT_NUM, DEFAULT_SP_BOLT_NUM);
    final int rcBoltNum = BenchmarkUtils.getInt(config, COUNTER_NUM, DEFAULT_RC_BOLT_NUM);
    final int windowLength = BenchmarkUtils.getInt(config, WINDOW_LENGTH,
            RollingBolt.DEFAULT_SLIDING_WINDOW_IN_SECONDS);
    final int emitFreq = BenchmarkUtils.getInt(config, EMIT_FREQ,
            RollingBolt.DEFAULT_EMIT_FREQUENCY_IN_SECONDS);

    spout = new FileReadSpout(BenchmarkUtils.ifAckEnabled(config));

    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout(SPOUT_ID, spout, spoutNum);
    builder.setBolt(SPLIT_ID, new WordCount.SplitSentence(), spBoltNum)
            .localOrShuffleGrouping(SPOUT_ID);
    builder.setBolt(COUNTER_ID, new RollingCountBolt(windowLength, emitFreq), rcBoltNum)
            .fieldsGrouping(SPLIT_ID, new Fields(WordCount.SplitSentence.FIELDS));
    return builder.createTopology();
  }

}