@Test public void testClose() throws Exception { BatchSpoutOutputCollector collectorMock = mock(BatchSpoutOutputCollector.class); PowerMockito.whenNew(BatchSpoutOutputCollector.class).withAnyArguments().thenReturn(collectorMock); SpoutOutputBatcher spout = new SpoutOutputBatcher(this.spoutMock, this.noBatching); spout.open(null, null, null); spout.close(); verify(this.spoutMock).close(); verify(collectorMock).flush(); }
@Override public void close() { this.wrappedSpout.close(); this.batchCollector.flush(); }
SpoutOutputBatcher implements IRichSpout { @Override public void close() { this.wrappedSpout.close(); this.batchCollector.flush(); } }
SpoutOutputBatcher implements IRichSpout { @Override public void close() { this.wrappedSpout.close(); this.batchCollector.flush(); } SpoutOutputBatcher(IRichSpout spout, int batchSize);  SpoutOutputBatcher(IRichSpout spout, Map<String, Integer> batchSizes); }
SpoutOutputBatcher implements IRichSpout { @Override public void close() { this.wrappedSpout.close(); this.batchCollector.flush(); } SpoutOutputBatcher(IRichSpout spout, int batchSize);  SpoutOutputBatcher(IRichSpout spout, Map<String, Integer> batchSizes); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void close(); @Override void activate(); @Override void deactivate(); @Override void nextTuple(); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration(); static void registerKryoClasses(Config stormConfig); }
SpoutOutputBatcher implements IRichSpout { @Override public void close() { this.wrappedSpout.close(); this.batchCollector.flush(); } SpoutOutputBatcher(IRichSpout spout, int batchSize);  SpoutOutputBatcher(IRichSpout spout, Map<String, Integer> batchSizes); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void close(); @Override void activate(); @Override void deactivate(); @Override void nextTuple(); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration(); static void registerKryoClasses(Config stormConfig);  }