ConfigurableIngestTopology { void configure(HierarchicalConfiguration conf) throws Exception { this.stormConfig = new Config(); this.builder = new TopologyBuilder(); SubnodeConfiguration topologyConf = conf.configurationAt(TOPOLOGY); this.topologyName = getTopologyName(topologyConf); SubnodeConfiguration stormConf = topologyConf.configurationAt(STORM_CONF); configureStorm(stormConf, stormConfig); SubnodeConfiguration spoutConf = conf.configurationAt(SPOUT); String spout = configureSpout(builder, spoutConf); SubnodeConfiguration boltConfs = conf.configurationAt(BOLTS); configureBolts(builder, boltConfs, spout); }  static void main(String[] args);  }
@Test public void testConfigure( @Injectable HierarchicalConfiguration conf, @Injectable SubnodeConfiguration topologyConf, @Injectable SubnodeConfiguration stormConf, @Injectable SubnodeConfiguration spoutConf, @Injectable SubnodeConfiguration boltConfs, @Injectable String topologyName, @Injectable String spout) throws Exception { new Expectations(topology) {{ conf.configurationAt(TOPOLOGY); result = topologyConf; topology.getTopologyName(topologyConf); result = topologyName; topologyConf.configurationAt(STORM_CONF); result = stormConf; topology.configureStorm(stormConf, (Config) any); conf.configurationAt(SPOUT); result = spoutConf; topology.configureSpout((TopologyBuilder) any, spoutConf); result = spout; conf.configurationAt(BOLTS); result = boltConfs; topology.configureBolts((TopologyBuilder) any, boltConfs, spout); }}; topology.configure(conf); }