SpoutWrapper extends RichParallelSourceFunction<OUT> implements StoppableFunction { @Override public final void run(final SourceContext<OUT> ctx) throws Exception { final GlobalJobParameters config = super.getRuntimeContext().getExecutionConfig() .getGlobalJobParameters(); StormConfig stormConfig = new StormConfig(); if (config != null) { if (config instanceof StormConfig) { stormConfig = (StormConfig) config; } else { stormConfig.putAll(config.toMap()); } } final TopologyContext stormTopologyContext = WrapperSetupHelper.createTopologyContext( (StreamingRuntimeContext) super.getRuntimeContext(), this.spout, this.name, this.stormTopology, stormConfig); SpoutCollector<OUT> collector = new SpoutCollector<OUT>(this.numberOfAttributes, stormTopologyContext.getThisTaskId(), ctx); this.spout.open(stormConfig, stormTopologyContext, new SpoutOutputCollector(collector)); this.spout.activate(); if (numberOfInvocations == null) { if (this.spout instanceof FiniteSpout) { final FiniteSpout finiteSpout = (FiniteSpout) this.spout; while (this.isRunning && !finiteSpout.reachedEnd()) { finiteSpout.nextTuple(); } } else { while (this.isRunning) { this.spout.nextTuple(); } } } else { int counter = this.numberOfInvocations; if (counter >= 0) { while ((--counter >= 0) && this.isRunning) { this.spout.nextTuple(); } } else { do { collector.tupleEmitted = false; this.spout.nextTuple(); } while (collector.tupleEmitted && this.isRunning); } } } 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();  }
@SuppressWarnings("unchecked") @Test public void testRunExecuteFixedNumber() throws Exception { final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer(); declarer.declare(new Fields("dummy")); PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments() .thenReturn(declarer); final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class); when(taskContext.getExecutionConfig()).thenReturn(mock(ExecutionConfig.class)); when(taskContext.getTaskName()).thenReturn("name"); final IRichSpout spout = mock(IRichSpout.class); final int numberOfCalls = this.r.nextInt(50); final SpoutWrapper<?> spoutWrapper = new SpoutWrapper<Object>(spout, numberOfCalls); spoutWrapper.setRuntimeContext(taskContext); spoutWrapper.run(mock(SourceContext.class)); verify(spout, times(numberOfCalls)).nextTuple(); }
@Test public void testRunExecuteFinite() throws Exception { final int numberOfCalls = this.r.nextInt(50); final LinkedList<Tuple1<Integer>> expectedResult = new LinkedList<Tuple1<Integer>>(); for (int i = numberOfCalls - 1; i >= 0; --i) { expectedResult.add(new Tuple1<Integer>(new Integer(i))); } final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class); when(taskContext.getExecutionConfig()).thenReturn(mock(ExecutionConfig.class)); when(taskContext.getTaskName()).thenReturn("name"); final FiniteTestSpout spout = new FiniteTestSpout(numberOfCalls); final SpoutWrapper<Tuple1<Integer>> spoutWrapper = new SpoutWrapper<Tuple1<Integer>>( spout, -1); spoutWrapper.setRuntimeContext(taskContext); final TestContext collector = new TestContext(); spoutWrapper.run(collector); Assert.assertEquals(expectedResult, collector.result); }
@SuppressWarnings("unchecked") @Test public void runAndExecuteFiniteSpout() throws Exception { final FiniteSpout stormSpout = mock(FiniteSpout.class); when(stormSpout.reachedEnd()).thenReturn(false, false, false, true, false, false, true); final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class); when(taskContext.getExecutionConfig()).thenReturn(mock(ExecutionConfig.class)); when(taskContext.getTaskName()).thenReturn("name"); final SpoutWrapper<?> wrapper = new SpoutWrapper<Object>(stormSpout); wrapper.setRuntimeContext(taskContext); wrapper.run(mock(SourceContext.class)); verify(stormSpout, times(3)).nextTuple(); }
@SuppressWarnings("unchecked") @Test public void runAndExecuteFiniteSpout2() throws Exception { final FiniteSpout stormSpout = mock(FiniteSpout.class); when(stormSpout.reachedEnd()).thenReturn(true, false, true, false, true, false, true); final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class); when(taskContext.getExecutionConfig()).thenReturn(mock(ExecutionConfig.class)); when(taskContext.getTaskName()).thenReturn("name"); final SpoutWrapper<?> wrapper = new SpoutWrapper<Object>(stormSpout); wrapper.setRuntimeContext(taskContext); wrapper.run(mock(SourceContext.class)); verify(stormSpout, never()).nextTuple(); }