class SOL extends StormBenchmark {

  @Override
  public StormTopology getTopology(Config config) {
    final int numLevels = BenchmarkUtils.getInt(config, TOPOLOGY_LEVEL, DEFAULT_NUM_LEVELS);
    final int msgSize = BenchmarkUtils.getInt(config, RandomMessageSpout.MESSAGE_SIZE,
            RandomMessageSpout.DEFAULT_MESSAGE_SIZE);
    final int spoutNum = BenchmarkUtils.getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
    final int boltNum = BenchmarkUtils.getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM);

    spout = new RandomMessageSpout(msgSize, BenchmarkUtils.ifAckEnabled(config));

    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout(SPOUT_ID, spout, spoutNum);
    builder.setBolt(BOLT_ID + 1, new ConstBolt(), boltNum)
        .shuffleGrouping(SPOUT_ID);
    for (int levelNum = 2; levelNum <= numLevels - 1; levelNum++) {
      builder.setBolt(BOLT_ID + levelNum, new ConstBolt(), boltNum)
        .shuffleGrouping(BOLT_ID + (levelNum - 1));
    }
   return builder.createTopology();
  }

}