ConfigurableIngestTopology { BoltDeclarer buildBoltDeclarer(TopologyBuilder builder, StormParallelismConfig stormParallelismConfig, String boltId, IComponent baseComponent) throws ConfigurationException { if (baseComponent instanceof IBasicBolt) { return builder .setBolt(boltId, (IBasicBolt) baseComponent, stormParallelismConfig.getParallelismHint()) .setNumTasks(stormParallelismConfig.getNumTasks()); } else if (baseComponent instanceof IRichBolt) { return builder .setBolt(boltId, (IRichBolt) baseComponent, stormParallelismConfig.getParallelismHint()) .setNumTasks(stormParallelismConfig.getNumTasks()); } else { throw new ConfigurationException("Unknown component type: " + baseComponent.getClass()); } }  static void main(String[] args);  }
@Test public void testBuildBoltDeclarerIBasicBolt( @Injectable TopologyBuilder builder, @Injectable StormParallelismConfig stormParallelismConfig, @Injectable String boltId, @Injectable IBasicBolt baseComponent, @Injectable BoltDeclarer boltDeclarer) throws Exception { new Expectations() {{ stormParallelismConfig.getParallelismHint(); result = 1; stormParallelismConfig.getNumTasks(); result = 1; builder.setBolt(boltId, baseComponent, 1); result = boltDeclarer; boltDeclarer.setNumTasks(1); }}; topology.buildBoltDeclarer(builder, stormParallelismConfig, boltId, baseComponent); }
@Test public void testBuildBoltDeclarerIRichBolt( @Injectable TopologyBuilder builder, @Injectable StormParallelismConfig stormParallelismConfig, @Injectable String boltId, @Injectable IRichBolt baseComponent, @Injectable BoltDeclarer boltDeclarer) throws Exception { new Expectations() {{ stormParallelismConfig.getParallelismHint(); result = 1; stormParallelismConfig.getNumTasks(); result = 1; builder.setBolt(boltId, baseComponent, 1); result = boltDeclarer; boltDeclarer.setNumTasks(1); }}; topology.buildBoltDeclarer(builder, stormParallelismConfig, boltId, baseComponent); }
@Test public void testBuildBoltDeclarerUnknownComponent( @Injectable TopologyBuilder builder, @Injectable StormParallelismConfig stormParallelismConfig, @Injectable String boltId, @Injectable TestClassUnknownComponent baseComponent) throws Exception { thrown.expect(ConfigurationException.class); topology.buildBoltDeclarer(builder, stormParallelismConfig, boltId, baseComponent); }