class BoltWrapper extends AbstractStreamOperator<OUT> implements OneInputStreamOperator<IN, OUT> {

	@Override
	public void open() throws Exception {
		super.open();

		this.flinkCollector = new TimestampedCollector<>(this.output);

		GlobalJobParameters config = getExecutionConfig().getGlobalJobParameters();
		StormConfig stormConfig = new StormConfig();

		if (config != null) {
			if (config instanceof StormConfig) {
				stormConfig = (StormConfig) config;
			} else {
				stormConfig.putAll(config.toMap());
			}
		}

		this.topologyContext = WrapperSetupHelper.createTopologyContext(
				getRuntimeContext(), this.bolt, this.name, this.stormTopology, stormConfig);

		final OutputCollector stormCollector = new OutputCollector(new BoltCollector<OUT>(
				this.numberOfAttributes, this.topologyContext.getThisTaskId(), this.flinkCollector));

		if (this.stormTopology != null) {
			Map<GlobalStreamId, Grouping> inputs = this.topologyContext.getThisSources();

			for (GlobalStreamId inputStream : inputs.keySet()) {
				for (Integer tid : this.topologyContext.getComponentTasks(inputStream
						.get_componentId())) {
					this.inputComponentIds.put(tid, inputStream.get_componentId());
					this.inputStreamIds.put(tid, inputStream.get_streamId());
					this.inputSchemas.put(tid,
							this.topologyContext.getComponentOutputFields(inputStream));
				}
			}
		}

		this.bolt.prepare(stormConfig, this.topologyContext, stormCollector);
	}

	public  BoltWrapper(final IRichBolt bolt);
	public  BoltWrapper(final IRichBolt bolt, final Fields inputSchema);
	public  BoltWrapper(final IRichBolt bolt, final String[] rawOutputs);
	public  BoltWrapper(final IRichBolt bolt, final Collection<String> rawOutputs);
	public  BoltWrapper(final IRichBolt bolt, final Fields inputSchema, final String[] rawOutputs);
	public  BoltWrapper(final IRichBolt bolt, final Fields inputSchema,
			final Collection<String> rawOutputs);
	public  BoltWrapper(final IRichBolt bolt, final String name, final String inputStreamId,
			final String inputComponentId, final Fields inputSchema,
			final Collection<String> rawOutputs);

}