class StreamingJobGraphGenerator {

	@VisibleForTesting
	static boolean areOperatorsChainable(
			StreamNode upStreamVertex,
			StreamNode downStreamVertex,
			StreamGraph streamGraph) {
		StreamOperatorFactory<?> upStreamOperator = upStreamVertex.getOperatorFactory();
		StreamOperatorFactory<?> downStreamOperator = downStreamVertex.getOperatorFactory();
		if (downStreamOperator == null || upStreamOperator == null) {
			return false;
		}

		// yielding operators cannot be chained to legacy sources
		// unfortunately the information that vertices have been chained is not preserved at this point
		if (downStreamOperator instanceof YieldingOperatorFactory &&
				getHeadOperator(upStreamVertex, streamGraph).isStreamSource()) {
			return false;
		}

		// we use switch/case here to make sure this is exhaustive if ever values are added to the
		// ChainingStrategy enum
		boolean isChainable;

		switch (upStreamOperator.getChainingStrategy()) {
			case NEVER:
				isChainable = false;
				break;
			case ALWAYS:
			case HEAD:
			case HEAD_WITH_SOURCES:
				isChainable = true;
				break;
			default:
				throw new RuntimeException("Unknown chaining strategy: " + upStreamOperator.getChainingStrategy());
		}

		switch (downStreamOperator.getChainingStrategy()) {
			case NEVER:
			case HEAD:
				isChainable = false;
				break;
			case ALWAYS:
				// keep the value from upstream
				break;
			case HEAD_WITH_SOURCES:
				// only if upstream is a source
				isChainable &= (upStreamOperator instanceof SourceOperatorFactory);
				break;
			default:
				throw new RuntimeException("Unknown chaining strategy: " + upStreamOperator.getChainingStrategy());
		}

		return isChainable;
	}

	private  StreamingJobGraphGenerator(StreamGraph streamGraph, @Nullable JobID jobID);

}