SpoutWrapper extends RichParallelSourceFunction<OUT> implements StoppableFunction { @Override public void cancel() { this.isRunning = false; } SpoutWrapper(final IRichSpout spout);  SpoutWrapper(final IRichSpout spout, final Integer numberOfInvocations);  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs);  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs,
			final Integer numberOfInvocations);  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs);  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs,
			final Integer numberOfInvocations);  SpoutWrapper(final IRichSpout spout, final String name, final Collection<String> rawOutputs,
			final Integer numberOfInvocations); void setStormTopology(StormTopology stormTopology); @Override final void run(final SourceContext<OUT> ctx); @Override void cancel(); @Override void stop(); @Override void close();  }
@Test public void testCancel() throws Exception { final int numberOfCalls = 5 + this.r.nextInt(5); final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class); when(taskContext.getExecutionConfig()).thenReturn(mock(ExecutionConfig.class)); when(taskContext.getTaskName()).thenReturn("name"); final IRichSpout spout = new FiniteTestSpout(numberOfCalls); final SpoutWrapper<Tuple1<Integer>> spoutWrapper = new SpoutWrapper<Tuple1<Integer>>(spout); spoutWrapper.setRuntimeContext(taskContext); spoutWrapper.cancel(); final TestContext collector = new TestContext(); spoutWrapper.run(collector); Assert.assertEquals(new LinkedList<Tuple1<Integer>>(), collector.result); }