class 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);
			}
		}
	}

	public  SpoutWrapper(final IRichSpout spout);
	public  SpoutWrapper(final IRichSpout spout, final Integer numberOfInvocations);
	public  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs);
	public  SpoutWrapper(final IRichSpout spout, final String[] rawOutputs,
			final Integer numberOfInvocations);
	public  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs);
	public  SpoutWrapper(final IRichSpout spout, final Collection<String> rawOutputs,
			final Integer numberOfInvocations);
	public  SpoutWrapper(final IRichSpout spout, final String name, final Collection<String> rawOutputs,
			final Integer numberOfInvocations);

}