Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3621#discussion_r108621295
--- Diff:
flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java
---
@@ -316,49 +341,45 @@ private void convertToSingletonState(final State<T>
sourceState, final State<T>
}
/**
- * Converts the given state into looping one. Looping state is
one with TAKE edge to itself and
+ * Creates the given state as a looping one. Looping state is
one with TAKE edge to itself and
* PROCEED edge to the sinkState. It also consists of a similar
state without the PROCEED edge, so that
* for each PROCEED transition branches in computation state
graph can be created only once.
*
- * <p>If this looping state is first of a graph we should treat
the {@link Pattern} as {@link FollowedByPattern}
- * to enable combinations.
- *
- * @param sourceState the state to converted
- * @param sinkState the state that the converted state
should point to
- * @param isFirstState if the looping state is first of a graph
+ * @param sinkState the state that the converted state should
point to
+ * @return the first state of the created complex state
*/
@SuppressWarnings("unchecked")
- private void convertToLooping(final State<T> sourceState, final
State<T> sinkState, boolean isFirstState) {
+ private State<T> createLooping(final State<T> sinkState) {
+ final State<T> loopingState = createNormalState();
final IterativeCondition<T> filterFunction =
(IterativeCondition<T>) currentPattern.getCondition();
- final IterativeCondition<T> trueFunction =
BooleanConditions.<T>trueFunction();
+ final IterativeCondition<T> trueFunction =
BooleanConditions.trueFunction();
- sourceState.addProceed(sinkState, trueFunction);
- sourceState.addTake(filterFunction);
- if (currentPattern instanceof FollowedByPattern ||
isFirstState) {
- final State<T> ignoreState = new State<>(
- currentPattern.getName(),
- State.StateType.Normal);
+ loopingState.addProceed(sinkState, trueFunction);
+ loopingState.addTake(filterFunction);
+ if
(!currentPattern.getQuantifier().hasProperty(QuantifierProperty.STRICT)) {
+ final State<T> ignoreState =
createNormalState();
final IterativeCondition<T> ignoreCondition =
getIgnoreCondition(currentPattern);
- sourceState.addIgnore(ignoreState,
ignoreCondition);
- ignoreState.addTake(sourceState,
filterFunction);
+ ignoreState.addTake(loopingState,
filterFunction);
ignoreState.addIgnore(ignoreState,
ignoreCondition);
--- End diff --
here you could remove the first argument and use the `void addIgnore(final
FilterFunction<T> condition)` right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
