ConfigurableIngestTopology { String configureSpout(TopologyBuilder builder, Configuration spout) throws ConfigurationException { String spoutType = spout.getString(TYPE); Configuration spoutConf = spout.subset(CONF); StormParallelismConfig parallelismConfig = getStormParallelismConfig(spoutConf); IRichSpout spoutComponent = (IRichSpout) buildComponent(spoutType, spoutConf); builder .setSpout(spoutType, spoutComponent, parallelismConfig.getParallelismHint()) .setNumTasks(parallelismConfig.getNumTasks()); return spoutType; }  static void main(String[] args);  }
@Test public void testConfigureSpout( @Injectable TopologyBuilder builder, @Injectable Configuration spout, @Injectable Configuration spoutConf, @Injectable StormParallelismConfig stormParallelismConfig, @Injectable IRichSpout spoutComponent) throws Exception { String spoutType = "spout-class"; new Expectations(topology) {{ spout.getString(TYPE); result = spoutType; spout.subset(CONF); result = spoutConf; topology.getStormParallelismConfig(spoutConf); result = stormParallelismConfig; topology.buildComponent(spoutType, spoutConf); result = spoutComponent; }}; assertThat(topology.configureSpout(builder, spout), is(spoutType)); }