flink cep源码分析

ignore: 忽略此次匹配的元素
proceed: 相当于forward的意思，到达下一个状态，不存储元素，继续做下一个状态的condition判断
take： 存储本次的元素

private State<T> createMiddleStates(final State<T> sinkState) {
			State<T> lastSink = sinkState;
			// 不断往上遍历pattern进行state的生成
			while (currentPattern.getPrevious() != null) {

				if (currentPattern.getQuantifier().getConsumingStrategy() == Quantifier.ConsumingStrategy.NOT_FOLLOW) {
					//skip notFollow patterns, they are converted into edge conditions
				} else if (currentPattern.getQuantifier().getConsumingStrategy() == Quantifier.ConsumingStrategy.NOT_NEXT) {
					final State<T> notNext = createState(currentPattern.getName(), State.StateType.Normal);
					final IterativeCondition<T> notCondition = getTakeCondition(currentPattern);
					// 否定类型的pattern需要创建一个stop state
					final State<T> stopState = createStopState(notCondition, currentPattern.getName());

					if (lastSink.isFinal()) {
						//so that the proceed to final is not fired
						结尾状态不用proceed过去做下一次计算了，可以直接ignore到Final，然后输出结果
						notNext.addIgnore(lastSink, new NotCondition<>(notCondition));
					} else {
						notNext.addProceed(lastSink, new NotCondition<>(notCondition));
					}
					// 在满足Not_NEXT的条件的时候就转化成stop状态即匹配失败
					notNext.addProceed(stopState, notCondition);
					lastSink = notNext;
				} else {
					// 非否定类型的状态的处理逻辑都在这个方法中
					lastSink = convertPattern(lastSink);
				}

				// we traverse the pattern graph backwards
				followingPattern = currentPattern;
				currentPattern = currentPattern.getPrevious();

				final Time currentWindowTime = currentPattern.getWindowTime();
				if (currentWindowTime != null && currentWindowTime.toMilliseconds() < windowTime) {
					// the window time is the global minimum of all window times of each state
					windowTime = currentWindowTime.toMilliseconds();
				}
			}
			return lastSink;
		}

protected void saveRegisterWatermarkTimer() {
	long currentWatermark = timerService.currentWatermark();
	// protect against overflow
	if (currentWatermark + 1 > currentWatermark) {
		timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, currentWatermark + 1);
	}
}

if (!computationState.isStartState() &&
	windowTime > 0L &&
	timestamp - computationState.getStartTimestamp() >= windowTime) {

	if (handleTimeout) {
		// extract the timed out event pattern
		Map<String, List<T>> timedOutPattern = extractCurrentMatches(computationState);
		timeoutResult.add(Tuple2.of(timedOutPattern, timestamp));
	}

	eventSharedBuffer.release(
			NFAStateNameHandler.getOriginalNameFromInternal(computationState.getPreviousState().getName()),
			computationState.getEvent(),
			computationState.getTimestamp(),
			computationState.getCounter());

	newComputationStates = Collections.emptyList();
	nfaChanged = true;
} else if (event != null) {
   // 在computeNextState的时候判断成功的take条件会将元素put到eventSharedBuffer中
	newComputationStates = computeNextStates(computationState, event, timestamp);

	if (newComputationStates.size() != 1) {
		nfaChanged = true;
	} else if (!newComputationStates.iterator().next().equals(computationState)) {
		nfaChanged = true;
	}
} else {
	newComputationStates = Collections.singleton(computationState);
}