NullTerminatingSpout implements FiniteSpout { @Override public boolean reachedEnd() { return this.observer.emitted == false; } NullTerminatingSpout(IRichSpout spout); @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(); @Override boolean reachedEnd();  }
@Test public void testReachedEnd() { NullTerminatingSpout finiteSpout = new NullTerminatingSpout(new TestDummySpout()); finiteSpout.open(null, null, mock(SpoutOutputCollector.class)); Assert.assertFalse(finiteSpout.reachedEnd()); finiteSpout.nextTuple(); Assert.assertFalse(finiteSpout.reachedEnd()); finiteSpout.nextTuple(); Assert.assertTrue(finiteSpout.reachedEnd()); }