StormProcessingItem extends AbstractProcessingItem implements StormTopologyNode { @Override public void addToTopology(StormTopology topology, int parallelismHint) { if(piBoltDeclarer != null){ }else{ TopologyBuilder stormBuilder = topology.getStormBuilder(); this.piBoltDeclarer = stormBuilder.setBolt(this.getName(), this.piBolt, parallelismHint); } } StormProcessingItem(Processor processor, int parallelismHint);   StormProcessingItem(Processor processor, String friendlyId, int parallelismHint); @Override void addToTopology(StormTopology topology, int parallelismHint); @Override StormStream createStream(); @Override String getId(); @Override String toString();  }
@Test public void testAddToTopology() { new Expectations() { { topology.getStormBuilder(); result = stormBuilder; stormBuilder.setBolt(ID, (IRichBolt) any, anyInt); result = new MockUp<BoltDeclarer>() { }.getMockInstance(); } }; pi.addToTopology(topology, PARRALLELISM_HINT_4); new Verifications() { { assertEquals(pi.getProcessor(), processor); assertEquals(pi.getParallelism(), PARRALLELISM_HINT_2); assertEquals(pi.getId(), ID); } }; }