ConfigurableIngestTopology { void configureBolts(TopologyBuilder builder, SubnodeConfiguration boltsConf, String spout) throws ConfigurationException { String prevComponent = spout; HashMap<Integer, String> boltNumberToId = new HashMap<>(); List<HierarchicalConfiguration> bolts = boltsConf.configurationsAt(BOLT); for (HierarchicalConfiguration bolt : bolts) { String boltType = bolt.getString(TYPE); Configuration boltConf = bolt.configurationAt(CONF); int boltNum = bolt.getInt(NUMBER_ATTRIBUTE); String boltId = String.format("%02d_%s", boltNum, boltType); boltNumberToId.put(boltNum, boltId); String subscribingToComponent = getSubscribingToComponent(prevComponent, boltNumberToId, boltConf); logger.info("{} subscribing to {}", boltId, subscribingToComponent); IComponent baseComponent = buildComponent(boltType, boltConf); StormParallelismConfig stormParallelismConfig = getStormParallelismConfig(boltConf); BoltDeclarer declarer = buildBoltDeclarer(builder, stormParallelismConfig, boltId, baseComponent); configureTickFrequency(boltConf, declarer); configureStreamGrouping(subscribingToComponent, boltConf, declarer); prevComponent = boltId; boltNum++; } }  static void main(String[] args);  }
@Test public void testConfigureBolts( @Injectable TopologyBuilder builder, @Injectable SubnodeConfiguration boltsConf, @Injectable HierarchicalConfiguration bolt, @Injectable SubnodeConfiguration boltConf, @Injectable IComponent baseComponent, @Injectable StormParallelismConfig stormParallelismConfig, @Injectable BoltDeclarer declarer) throws Exception { List<HierarchicalConfiguration> bolts = Arrays.asList(bolt); String spout = "spout_id"; new Expectations(topology) {{ boltsConf.configurationsAt(BOLT); result = bolts; bolt.getString(TYPE); result = "bolt_type"; bolt.configurationAt(CONF); result = boltConf; bolt.getInt(NUMBER_ATTRIBUTE); result = 0; topology.getSubscribingToComponent(spout, (HashMap<Integer, String>) any, boltConf); result = spout; topology.buildComponent("bolt_type", boltConf); result = baseComponent; topology.getStormParallelismConfig(boltConf); result = stormParallelismConfig; topology.buildBoltDeclarer(builder, stormParallelismConfig, "00_bolt_type", baseComponent); result = declarer; topology.configureTickFrequency(boltConf, declarer); topology.configureStreamGrouping(spout, boltConf, declarer); }}; topology.configureBolts(builder, boltsConf, spout); }