test_class.file: flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/SpoutWrapperTest.java
test_class.identifier: SpoutWrapperTest
test_class.superclass: extends AbstractTest
test_case.body: @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();
  	}
test_case.class_method_signature: SpoutWrapperTest.runAndExecuteFiniteSpout2()
test_case.full_signature: @SuppressWarnings("unchecked") @Test public void runAndExecuteFiniteSpout2()
test_case.identifier: runAndExecuteFiniteSpout2
test_case.invocations: mock thenReturn when reachedEnd mock thenReturn when getExecutionConfig mock thenReturn when getTaskName setRuntimeContext run mock nextTuple verify never
test_case.modifiers: @SuppressWarnings("unchecked") @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void runAndExecuteFiniteSpout2()
focal_class.declarator: serialVersionUID = -218340336648247605L
focal_class.modifier: private static final
focal_class.original_string: private static final long serialVersionUID = -218340336648247605L;
focal_class.type: long
focal_class.var_name: serialVersionUID
focal_class.declarator: numberOfAttributes
focal_class.modifier: private final
focal_class.original_string: private final HashMap<String, Integer> numberOfAttributes;
focal_class.type: HashMap<String, Integer>
focal_class.var_name: numberOfAttributes
focal_class.declarator: spout
focal_class.modifier: private final
focal_class.original_string: private final IRichSpout spout;
focal_class.type: IRichSpout
focal_class.var_name: spout
focal_class.declarator: name
focal_class.modifier: private final
focal_class.original_string: private final String name;
focal_class.type: String
focal_class.var_name: name
focal_class.declarator: isRunning = true
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean isRunning = true;
focal_class.type: boolean
focal_class.var_name: isRunning
focal_class.declarator: numberOfInvocations
focal_class.modifier: private
focal_class.original_string: private Integer numberOfInvocations;
focal_class.type: Integer
focal_class.var_name: numberOfInvocations
focal_class.declarator: stormTopology
focal_class.modifier: private
focal_class.original_string: private StormTopology stormTopology;
focal_class.type: StormTopology
focal_class.var_name: stormTopology
focal_class.file: flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/SpoutWrapper.java
focal_class.identifier: SpoutWrapper
focal_class.interfaces: implements StoppableFunction
focal_class.class_method_signature: SpoutWrapper.SpoutWrapper(final IRichSpout spout)
focal_class.full_signature: public  SpoutWrapper(final IRichSpout spout)
focal_class.identifier: SpoutWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichSpout spout)
focal_class.signature:  SpoutWrapper(final IRichSpout spout)
focal_class.class_method_signature: SpoutWrapper.SpoutWrapper(final IRichSpout spout, final Integer numberOfInvocations)
focal_class.full_signature: public  SpoutWrapper(final IRichSpout spout, final Integer numberOfInvocations)
focal_class.identifier: SpoutWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichSpout spout, final Integer numberOfInvocations)
focal_class.signature:  SpoutWrapper(final IRichSpout spout, final Integer numberOfInvocations)
focal_class.class_method_signature: SpoutWrapper.SpoutWrapper(final IRichSpout spout, final String[] rawOutputs)
focal_class.full_signature: public  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs)
focal_class.identifier: SpoutWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichSpout spout, final String[] rawOutputs)
focal_class.signature:  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs)
focal_class.class_method_signature: SpoutWrapper.SpoutWrapper(final IRichSpout spout, final String[] rawOutputs,
  			final Integer numberOfInvocations)
focal_class.full_signature: public  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs,
  			final Integer numberOfInvocations)
focal_class.identifier: SpoutWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichSpout spout, final String[] rawOutputs,
  			final Integer numberOfInvocations)
focal_class.signature:  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs,
  			final Integer numberOfInvocations)
focal_class.class_method_signature: SpoutWrapper.SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs)
focal_class.full_signature: public  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs)
focal_class.identifier: SpoutWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichSpout spout, final Collection<String> rawOutputs)
focal_class.signature:  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs)
focal_class.class_method_signature: SpoutWrapper.SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.full_signature: public  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.identifier: SpoutWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichSpout spout, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.signature:  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.class_method_signature: SpoutWrapper.SpoutWrapper(final IRichSpout spout, final String name, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.full_signature: public  SpoutWrapper(final IRichSpout spout, final String name, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.identifier: SpoutWrapper
focal_class.modifiers: public
focal_class.parameters: (final IRichSpout spout, final String name, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.signature:  SpoutWrapper(final IRichSpout spout, final String name, final Collection<String> rawOutputs,
  			final Integer numberOfInvocations)
focal_class.class_method_signature: SpoutWrapper.setStormTopology(StormTopology stormTopology)
focal_class.full_signature: public void setStormTopology(StormTopology stormTopology)
focal_class.identifier: setStormTopology
focal_class.modifiers: public
focal_class.parameters: (StormTopology stormTopology)
focal_class.return: void
focal_class.signature: void setStormTopology(StormTopology stormTopology)
focal_class.class_method_signature: SpoutWrapper.run(final SourceContext<OUT> ctx)
focal_class.full_signature: @Override public final void run(final SourceContext<OUT> ctx)
focal_class.identifier: run
focal_class.modifiers: @Override public final
focal_class.parameters: (final SourceContext<OUT> ctx)
focal_class.return: void
focal_class.signature: void run(final SourceContext<OUT> ctx)
focal_class.class_method_signature: SpoutWrapper.cancel()
focal_class.full_signature: @Override public void cancel()
focal_class.identifier: cancel
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void cancel()
focal_class.class_method_signature: SpoutWrapper.stop()
focal_class.full_signature: @Override public void stop()
focal_class.identifier: stop
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void stop()
focal_class.class_method_signature: SpoutWrapper.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.superclass: extends RichParallelSourceFunction<OUT>
focal_method.body: @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);
  			}
  		}
  	}
focal_method.class_method_signature: SpoutWrapper.run(final SourceContext<OUT> ctx)
focal_method.full_signature: @Override public final void run(final SourceContext<OUT> ctx)
focal_method.identifier: run
focal_method.invocations: getGlobalJobParameters getExecutionConfig getRuntimeContext putAll toMap createTopologyContext getRuntimeContext getThisTaskId open activate reachedEnd nextTuple nextTuple nextTuple nextTuple
focal_method.modifiers: @Override public final
focal_method.parameters: (final SourceContext<OUT> ctx)
focal_method.return: void
focal_method.signature: void run(final SourceContext<OUT> ctx)