class SourceOperator extends AbstractStreamOperator<OUT> implements OperatorEventHandler, PushingAsyncDataInput<OUT> {

	@Override
	public void open() throws Exception {
		final MetricGroup metricGroup = getMetricGroup();

		final SourceReaderContext context = new SourceReaderContext() {
			@Override
			public MetricGroup metricGroup() {
				return metricGroup;
			}

			@Override
			public Configuration getConfiguration() {
				return configuration;
			}

			@Override
			public String getLocalHostName() {
				return localHostname;
			}

			@Override
			public void sendSourceEventToCoordinator(SourceEvent event) {
				operatorEventGateway.sendEventToCoordinator(new SourceEventWrapper(event));
			}
		};

		// in the future when we support both batch and streaming modes for the source operator,
		// and when this one is migrated to the "eager initialization" operator (StreamOperatorV2),
		// then we should evaluate this during operator construction.
		eventTimeLogic = TimestampsAndWatermarks.createStreamingEventTimeLogic(
				watermarkStrategy,
				metricGroup,
				getProcessingTimeService(),
				getExecutionConfig().getAutoWatermarkInterval());

		sourceReader = readerFactory.apply(context);

		// restore the state if necessary.
		final List<SplitT> splits = CollectionUtil.iterableToList(readerState.get());
		if (!splits.isEmpty()) {
			sourceReader.addSplits(splits);
		}

		// Register the reader to the coordinator.
		registerReader();

		// Start the reader after registration, sending messages in start is allowed.
		sourceReader.start();

		eventTimeLogic.startPeriodicWatermarkEmits();
	}

	public  SourceOperator(
			Function<SourceReaderContext, SourceReader<OUT, SplitT>> readerFactory,
			OperatorEventGateway operatorEventGateway,
			SimpleVersionedSerializer<SplitT> splitSerializer,
			WatermarkStrategy<OUT> watermarkStrategy,
			ProcessingTimeService timeService,
			Configuration configuration,
			String localHostname);

}