import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
@Test
public void testNoUnnecessaryStateCopiesCreated() {
	final Pattern<Event, Event> pattern = Pattern.<Event>begin("start").where(startFilter)
		.notFollowedBy("not").where(startFilter)
		.followedBy("oneOrMore").where(startFilter).oneOrMore()
		.followedBy("end").where(endFilter);

	final NFACompiler.NFAFactoryCompiler<Event> nfaFactoryCompiler = new NFACompiler.NFAFactoryCompiler<>(pattern);
	nfaFactoryCompiler.compileFactory();

	int endStateCount = 0;
	for (State<Event> state : nfaFactoryCompiler.getStates()) {
		if (state.getName().equals("end")) {
			endStateCount++;
		}
	}

	assertEquals(1, endStateCount);
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	Properties properties = new Properties();
	properties.setProperty("bootstrap.servers", "localhost:9092");
	properties.setProperty("group.id", "test");

	DataStream<TemperatureEvent> inputEventStream = env.addSource(
			new FlinkKafkaConsumer09<TemperatureEvent>("test", new EventDeserializationSchema(), properties));

	Pattern<TemperatureEvent, ?> warningPattern = Pattern.<TemperatureEvent> begin("first")
			.subtype(TemperatureEvent.class).where(new FilterFunction<TemperatureEvent>() {
				private static final long serialVersionUID = 1L;

				public boolean filter(TemperatureEvent value) {
					if (value.getTemperature() >= 26.0) {
						return true;
					}
					return false;
				}
			}).within(Time.seconds(10));

	DataStream<Alert> patternStream = CEP.pattern(inputEventStream, warningPattern)
			.select(new PatternSelectFunction<TemperatureEvent, Alert>() {
				private static final long serialVersionUID = 1L;

				public Alert select(Map<String, TemperatureEvent> event) throws Exception {

					return new Alert("Temperature Rise Detected:" + event.get("first").getTemperature()
							+ " on machine name:" + event.get("first").getMachineName());
				}

			});

	patternStream.print();
	env.execute("CEP on Temperature Sensor");
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Tests that a Java8 lambda can be passed as a CEP select function.
 */
@Test
public void testLambdaSelectFunction() {
	TypeInformation<EventA> eventTypeInformation = TypeExtractor.getForClass(EventA.class);
	TypeInformation<EventB> outputTypeInformation = TypeExtractor.getForClass(EventB.class);

	DataStream<EventA> inputStream = new DataStream<>(
		StreamExecutionEnvironment.getExecutionEnvironment(),
		new SourceTransformation<>(
			"source",
			null,
			eventTypeInformation,
			1));

	Pattern<EventA, ?> dummyPattern = Pattern.begin("start");

	PatternStream<EventA> patternStream = new PatternStream<>(inputStream, dummyPattern);

	DataStream<EventB> result = patternStream.select(
			(Map<String, List<EventA>> map) -> new EventB()
	);

	assertEquals(outputTypeInformation, result.getType());
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Tests that a Java8 lambda can be passed as a CEP flat select function.
 */
@Test
public void testLambdaFlatSelectFunction() {
	TypeInformation<EventA> eventTypeInformation = TypeExtractor.getForClass(EventA.class);
	TypeInformation<EventB> outputTypeInformation = TypeExtractor.getForClass(EventB.class);

	DataStream<EventA> inputStream = new DataStream<>(
		StreamExecutionEnvironment.getExecutionEnvironment(),
		new SourceTransformation<>(
			"source",
			null,
			eventTypeInformation,
			1));

	Pattern<EventA, ?> dummyPattern = Pattern.begin("start");

	PatternStream<EventA> patternStream = new PatternStream<>(inputStream, dummyPattern);

	DataStream<EventB> result = patternStream.flatSelect(
		(Map<String, List<EventA>> map, Collector<EventB> collector) -> collector.collect(new EventB())
	);

	assertEquals(outputTypeInformation, result.getType());
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Check pattern after match skip strategy.
 */
private void checkPatternSkipStrategy() {
	if (afterMatchSkipStrategy.getStrategy() == AfterMatchSkipStrategy.SkipStrategy.SKIP_TO_FIRST ||
		afterMatchSkipStrategy.getStrategy() == AfterMatchSkipStrategy.SkipStrategy.SKIP_TO_LAST) {
		Pattern<T, ?> pattern = currentPattern;
		while (pattern.getPrevious() != null && !pattern.getName().equals(afterMatchSkipStrategy.getPatternName())) {
			pattern = pattern.getPrevious();
		}

		// pattern name match check.
		if (!pattern.getName().equals(afterMatchSkipStrategy.getPatternName())) {
			throw new MalformedPatternException("The pattern name specified in AfterMatchSkipStrategy " +
				"can not be found in the given Pattern");
		}
	}
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Retrieves list of conditions resulting in Stop state and names of the corresponding NOT patterns.
 *
 * <p>A current not condition can be produced in two cases:
 * <ol>
 *     <li>the previous pattern is a {@link Quantifier.ConsumingStrategy#NOT_FOLLOW}</li>
 *     <li>exists a backward path of {@link Quantifier.QuantifierProperty#OPTIONAL} patterns to
 *       {@link Quantifier.ConsumingStrategy#NOT_FOLLOW}</li>
 * </ol>
 *
 * <p><b>WARNING:</b> for more info on the second case see: {@link NFAFactoryCompiler#copyWithoutTransitiveNots(State)}
 *
 * @return list of not conditions with corresponding names
 */
private List<Tuple2<IterativeCondition<T>, String>> getCurrentNotCondition() {
	List<Tuple2<IterativeCondition<T>, String>> notConditions = new ArrayList<>();

	Pattern<T, ? extends T> previousPattern = currentPattern;
	while (previousPattern.getPrevious() != null && (
		previousPattern.getPrevious().getQuantifier().hasProperty(Quantifier.QuantifierProperty.OPTIONAL) ||
		previousPattern.getPrevious().getQuantifier().getConsumingStrategy() == Quantifier.ConsumingStrategy.NOT_FOLLOW)) {

		previousPattern = previousPattern.getPrevious();

		if (previousPattern.getQuantifier().getConsumingStrategy() == Quantifier.ConsumingStrategy.NOT_FOLLOW) {
			final IterativeCondition<T> notCondition = getTakeCondition(previousPattern);
			notConditions.add(Tuple2.of(notCondition, previousPattern.getName()));
		}
	}
	return notConditions;
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Create the states for the group pattern as a looping one.
 *
 * @param groupPattern the group pattern to create the states for
 * @param sinkState the state that the group pattern being converted should point to
 * @return the first state of the states of the group pattern
 */
private State<T> createLoopingGroupPatternState(
	final GroupPattern<T, ?> groupPattern,
	final State<T> sinkState) {
	final IterativeCondition<T> proceedCondition = getTrueFunction();

	Pattern<T, ?> oldCurrentPattern = currentPattern;
	Pattern<T, ?> oldFollowingPattern = followingPattern;
	GroupPattern<T, ?> oldGroupPattern = currentGroupPattern;

	final State<T> dummyState = createState(currentPattern.getName(), State.StateType.Normal);
	State<T> lastSink = dummyState;
	currentGroupPattern = groupPattern;
	currentPattern = groupPattern.getRawPattern();
	lastSink = createMiddleStates(lastSink);
	lastSink = convertPattern(lastSink);
	lastSink.addProceed(sinkState, proceedCondition);
	dummyState.addProceed(lastSink, proceedCondition);
	currentPattern = oldCurrentPattern;
	followingPattern = oldFollowingPattern;
	currentGroupPattern = oldGroupPattern;
	return lastSink;
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
private void testStartWithOneOrZeroOrMoreStrict(Pattern<Event, ?> pattern) {
	List<StreamRecord<Event>> inputEvents = new ArrayList<>();

	inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
	inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent1, 3));
	inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 4));
	inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent2, 5));
	inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 6));

	NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);

	List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);

	compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
		Lists.newArrayList(ConsecutiveData.middleEvent1),
		Lists.newArrayList(ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3),
		Lists.newArrayList(ConsecutiveData.middleEvent2),
		Lists.newArrayList(ConsecutiveData.middleEvent3)
	));
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	DataStream<TemperatureEvent> inputEventStream = env.fromElements(new TemperatureEvent("xyz", 22.0),
			new TemperatureEvent("xyz", 20.1), new TemperatureEvent("xyz", 21.1), new TemperatureEvent("xyz", 22.2),
			new TemperatureEvent("xyz", 29.1), new TemperatureEvent("xyz", 22.3), new TemperatureEvent("xyz", 22.1),
			new TemperatureEvent("xyz", 22.4), new TemperatureEvent("xyz", 22.7),
			new TemperatureEvent("xyz", 27.0));

	Pattern<TemperatureEvent, ?> warningPattern = Pattern.<TemperatureEvent> begin("first")
			.subtype(TemperatureEvent.class).where(new FilterFunction<TemperatureEvent>() {
				private static final long serialVersionUID = 1L;

				public boolean filter(TemperatureEvent value) {
					if (value.getTemperature() >= 26.0) {
						return true;
					}
					return false;
				}
			}).within(Time.seconds(10));

	DataStream<Alert> patternStream = CEP.pattern(inputEventStream, warningPattern)
			.select(new PatternSelectFunction<TemperatureEvent, Alert>() {
				private static final long serialVersionUID = 1L;

				public Alert select(Map<String, TemperatureEvent> event) throws Exception {

					return new Alert("Temperature Rise Detected");
				}

			});

	patternStream.print();
	env.execute("CEP on Temperature Sensor");
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
@Override
public Pattern<LocalWeatherData, ?> getEventPattern() {
    return Pattern
            .<LocalWeatherData>begin("First Event").where(
                    new SimpleCondition<LocalWeatherData>() {
                        @Override
                        public boolean filter(LocalWeatherData event) throws Exception {
                            return event.getWindSpeed() > 110;
                        }
                    });
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
@Override
public Pattern<LocalWeatherData, ?> getEventPattern() {
    return Pattern
            .<LocalWeatherData>begin("First Event").where(
                    new SimpleCondition<LocalWeatherData>() {
                        @Override
                        public boolean filter(LocalWeatherData event) throws Exception {
                            return event.getWindSpeed() >= 39 && event.getWindSpeed() <= 110;
                        }
                    });
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Compiles the given pattern into a {@link NFAFactory}. The NFA factory can be used to create
 * multiple NFAs.
 *
 * @param pattern Definition of sequence pattern
 * @param inputTypeSerializer Serializer for the input type
 * @param timeoutHandling True if the NFA shall return timed out event patterns
 * @param <T> Type of the input events
 * @return Factory for NFAs corresponding to the given pattern
 */
@SuppressWarnings("unchecked")
public static <T> NFAFactory<T> compileFactory(
	final Pattern<T, ?> pattern,
	final TypeSerializer<T> inputTypeSerializer,
	boolean timeoutHandling) {
	if (pattern == null) {
		// return a factory for empty NFAs
		return new NFAFactoryImpl<>(inputTypeSerializer, 0, Collections.<State<T>>emptyList(), timeoutHandling);
	} else {
		final NFAFactoryCompiler<T> nfaFactoryCompiler = new NFAFactoryCompiler<>(pattern);
		nfaFactoryCompiler.compileFactory();
		return new NFAFactoryImpl<>(inputTypeSerializer, nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling);
	}
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Check if there are duplicate pattern names. If yes, it
 * throws a {@link MalformedPatternException}.
 */
private void checkPatternNameUniqueness() {
	// make sure there is no pattern with name "$endState$"
	stateNameHandler.checkNameUniqueness(ENDING_STATE_NAME);
	Pattern patternToCheck = currentPattern;
	while (patternToCheck != null) {
		checkPatternNameUniqueness(patternToCheck);
		patternToCheck = patternToCheck.getPrevious();
	}
	stateNameHandler.clear();
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Checks if the given pattern is optional. If the given pattern is the head of a group pattern,
 * the optional status depends on the group pattern.
 */
private boolean isPatternOptional(Pattern<T, ?> pattern) {
	if (headOfGroup(pattern)) {
		return isCurrentGroupPatternFirstOfLoop() &&
			currentGroupPattern.getQuantifier().hasProperty(Quantifier.QuantifierProperty.OPTIONAL);
	} else {
		return pattern.getQuantifier().hasProperty(Quantifier.QuantifierProperty.OPTIONAL);
	}
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * Create all the states for the group pattern.
 *
 * @param groupPattern the group pattern to create the states for
 * @param sinkState the state that the group pattern being converted should point to
 * @param proceedState the state that the group pattern being converted should proceed to
 * @param isOptional whether the group pattern being converted is optional
 * @return the first state of the states of the group pattern
 */
private State<T> createGroupPatternState(
	final GroupPattern<T, ?> groupPattern,
	final State<T> sinkState,
	final State<T> proceedState,
	final boolean isOptional) {
	final IterativeCondition<T> proceedCondition = getTrueFunction();

	Pattern<T, ?> oldCurrentPattern = currentPattern;
	Pattern<T, ?> oldFollowingPattern = followingPattern;
	GroupPattern<T, ?> oldGroupPattern = currentGroupPattern;

	State<T> lastSink = sinkState;
	currentGroupPattern = groupPattern;
	currentPattern = groupPattern.getRawPattern();
	lastSink = createMiddleStates(lastSink);
	lastSink = convertPattern(lastSink);
	if (isOptional) {
		// for the first state of a group pattern, its PROCEED edge should point to
		// the following state of that group pattern
		lastSink.addProceed(proceedState, proceedCondition);
	}
	currentPattern = oldCurrentPattern;
	followingPattern = oldFollowingPattern;
	currentGroupPattern = oldGroupPattern;
	return lastSink;
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * @return The {@link IterativeCondition condition} for the {@code IGNORE} edge
 * that corresponds to the specified {@link Pattern} and extended with stop(until) condition
 * if necessary. It is applicable only for inner states of a complex state like looping or times.
 */
@SuppressWarnings("unchecked")
private IterativeCondition<T> getInnerIgnoreCondition(Pattern<T, ?> pattern) {
	Quantifier.ConsumingStrategy consumingStrategy = pattern.getQuantifier().getInnerConsumingStrategy();
	if (headOfGroup(pattern)) {
		// for the head pattern of a group pattern, we should consider the
		// inner consume strategy of the group pattern
		consumingStrategy = currentGroupPattern.getQuantifier().getInnerConsumingStrategy();
	}

	IterativeCondition<T> innerIgnoreCondition = null;
	switch (consumingStrategy) {
		case STRICT:
			innerIgnoreCondition = null;
			break;
		case SKIP_TILL_NEXT:
			innerIgnoreCondition = new NotCondition<>((IterativeCondition<T>) pattern.getCondition());
			break;
		case SKIP_TILL_ANY:
			innerIgnoreCondition = BooleanConditions.trueFunction();
			break;
	}

	if (currentGroupPattern != null && currentGroupPattern.getUntilCondition() != null) {
		innerIgnoreCondition = extendWithUntilCondition(
			innerIgnoreCondition,
			(IterativeCondition<T>) currentGroupPattern.getUntilCondition(),
			false);
	}
	return innerIgnoreCondition;
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * @return The {@link IterativeCondition condition} for the {@code IGNORE} edge
 * that corresponds to the specified {@link Pattern} and extended with
 * stop(until) condition if necessary. For more on strategy see {@link Quantifier}
 */
@SuppressWarnings("unchecked")
private IterativeCondition<T> getIgnoreCondition(Pattern<T, ?> pattern) {
	Quantifier.ConsumingStrategy consumingStrategy = pattern.getQuantifier().getConsumingStrategy();
	if (headOfGroup(pattern)) {
		// for the head pattern of a group pattern, we should consider the inner consume strategy
		// of the group pattern if the group pattern is not the head of the TIMES/LOOPING quantifier;
		// otherwise, we should consider the consume strategy of the group pattern
		if (isCurrentGroupPatternFirstOfLoop()) {
			consumingStrategy = currentGroupPattern.getQuantifier().getConsumingStrategy();
		} else {
			consumingStrategy = currentGroupPattern.getQuantifier().getInnerConsumingStrategy();
		}
	}

	IterativeCondition<T> ignoreCondition = null;
	switch (consumingStrategy) {
		case STRICT:
			ignoreCondition = null;
			break;
		case SKIP_TILL_NEXT:
			ignoreCondition = new NotCondition<>((IterativeCondition<T>) pattern.getCondition());
			break;
		case SKIP_TILL_ANY:
			ignoreCondition = BooleanConditions.trueFunction();
			break;
	}

	if (currentGroupPattern != null && currentGroupPattern.getUntilCondition() != null) {
		ignoreCondition = extendWithUntilCondition(
			ignoreCondition,
			(IterativeCondition<T>) currentGroupPattern.getUntilCondition(),
			false);
	}
	return ignoreCondition;
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
/**
 * @return the {@link IterativeCondition condition} for the {@code TAKE} edge
 * that corresponds to the specified {@link Pattern} and extended with
 * stop(until) condition if necessary.
 */
@SuppressWarnings("unchecked")
private IterativeCondition<T> getTakeCondition(Pattern<T, ?> pattern) {
	IterativeCondition<T> takeCondition = (IterativeCondition<T>) pattern.getCondition();
	if (currentGroupPattern != null && currentGroupPattern.getUntilCondition() != null) {
		takeCondition = extendWithUntilCondition(
			takeCondition,
			(IterativeCondition<T>) currentGroupPattern.getUntilCondition(),
			true);
	}
	return takeCondition;
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
@Override
public NFA<Event> createNFA() {

	Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new StartFilter())
			.followedBy("middle").subtype(SubEvent.class).where(new MiddleFilter())
			.followedBy("end").where(new EndFilter())
			// add a window timeout to test whether timestamps of elements in the
			// priority queue in CEP operator are correctly checkpointed/restored
			.within(Time.milliseconds(10L));

	return NFACompiler.compile(pattern, Event.createTypeSerializer(), handleTimeout);
}
 

import org.apache.flink.cep.pattern.Pattern; //导入依赖的package包/类
@Override
public NFA<Event> createNFA() {

	Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new StartFilter())
			.within(Time.milliseconds(10L));

	return NFACompiler.compile(pattern, Event.createTypeSerializer(), handleTimeout);
}