You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2017/04/28 12:30:50 UTC
[1/2] flink git commit: [FLINK-6208] [cep] Implement skip till next
match strategy
Repository: flink
Updated Branches:
refs/heads/master b527582d0 -> 7c35dc0e4
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
index f4bf647..4a00c1e 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.cep.Event;
import org.apache.flink.cep.SubEvent;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
import org.apache.flink.cep.pattern.Pattern;
+import org.apache.flink.cep.pattern.Quantifier;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.windowing.time.Time;
@@ -46,6 +47,7 @@ import java.util.Set;
import static org.junit.Assert.assertEquals;
+@SuppressWarnings("unchecked")
public class NFAITCase extends TestLogger {
@Test
@@ -56,12 +58,12 @@ public class NFAITCase extends TestLogger {
SubEvent middleEvent = new SubEvent(42, "foo", 1.0, 10.0);
Event endEvent = new Event(43, "end", 1.0);
- inputEvents.add(new StreamRecord<Event>(startEvent, 1));
- inputEvents.add(new StreamRecord<Event>(new Event(43, "foobar", 1.0), 2));
+ inputEvents.add(new StreamRecord<>(startEvent, 1));
+ inputEvents.add(new StreamRecord<>(new Event(43, "foobar", 1.0), 2));
inputEvents.add(new StreamRecord<Event>(new SubEvent(41, "barfoo", 1.0, 5.0), 3));
inputEvents.add(new StreamRecord<Event>(middleEvent, 3));
- inputEvents.add(new StreamRecord<Event>(new Event(43, "start", 1.0), 4));
- inputEvents.add(new StreamRecord<Event>(endEvent, 5));
+ inputEvents.add(new StreamRecord<>(new Event(43, "start", 1.0), 4));
+ inputEvents.add(new StreamRecord<>(endEvent, 5));
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@@ -212,12 +214,12 @@ public class NFAITCase extends TestLogger {
final Event middleEvent;
final Event endEvent;
- events.add(new StreamRecord<Event>(new Event(1, "start", 1.0), 1));
- events.add(new StreamRecord<Event>(startEvent = new Event(2, "start", 1.0), 2));
- events.add(new StreamRecord<Event>(middleEvent = new Event(3, "middle", 1.0), 3));
- events.add(new StreamRecord<Event>(new Event(4, "foobar", 1.0), 4));
- events.add(new StreamRecord<Event>(endEvent = new Event(5, "end", 1.0), 11));
- events.add(new StreamRecord<Event>(new Event(6, "end", 1.0), 13));
+ events.add(new StreamRecord<>(new Event(1, "start", 1.0), 1));
+ events.add(new StreamRecord<>(startEvent = new Event(2, "start", 1.0), 2));
+ events.add(new StreamRecord<>(middleEvent = new Event(3, "middle", 1.0), 3));
+ events.add(new StreamRecord<>(new Event(4, "foobar", 1.0), 4));
+ events.add(new StreamRecord<>(endEvent = new Event(5, "end", 1.0), 11));
+ events.add(new StreamRecord<>(new Event(6, "end", 1.0), 13));
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
@@ -273,12 +275,12 @@ public class NFAITCase extends TestLogger {
Set<Tuple2<Map<String, Event>, Long>> resultingTimeoutPatterns = new HashSet<>();
Set<Tuple2<Map<String, Event>, Long>> expectedTimeoutPatterns = new HashSet<>();
- events.add(new StreamRecord<Event>(new Event(1, "start", 1.0), 1));
- events.add(new StreamRecord<Event>(new Event(2, "start", 1.0), 2));
- events.add(new StreamRecord<Event>(new Event(3, "middle", 1.0), 3));
- events.add(new StreamRecord<Event>(new Event(4, "foobar", 1.0), 4));
- events.add(new StreamRecord<Event>(new Event(5, "end", 1.0), 11));
- events.add(new StreamRecord<Event>(new Event(6, "end", 1.0), 13));
+ events.add(new StreamRecord<>(new Event(1, "start", 1.0), 1));
+ events.add(new StreamRecord<>(new Event(2, "start", 1.0), 2));
+ events.add(new StreamRecord<>(new Event(3, "middle", 1.0), 3));
+ events.add(new StreamRecord<>(new Event(4, "foobar", 1.0), 4));
+ events.add(new StreamRecord<>(new Event(5, "end", 1.0), 11));
+ events.add(new StreamRecord<>(new Event(6, "end", 1.0), 13));
Map<String, Event> timeoutPattern1 = new HashMap<>();
timeoutPattern1.put("start", new Event(1, "start", 1.0));
@@ -306,14 +308,14 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = -3268741540234334074L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("middle");
}
- }).followedBy("end").where(new SimpleCondition<Event>() {
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = -8995174172182138608L;
@Override
@@ -353,13 +355,13 @@ public class NFAITCase extends TestLogger {
SubEvent nextOne2 = new SubEvent(45, "next-one", 1.0, 2.0);
Event endEvent= new Event(46, "end", 1.0);
- inputEvents.add(new StreamRecord<Event>(startEvent, 1));
+ inputEvents.add(new StreamRecord<>(startEvent, 1));
inputEvents.add(new StreamRecord<Event>(middleEvent1, 3));
inputEvents.add(new StreamRecord<Event>(middleEvent2, 4));
inputEvents.add(new StreamRecord<Event>(middleEvent3, 5));
inputEvents.add(new StreamRecord<Event>(nextOne1, 6));
inputEvents.add(new StreamRecord<Event>(nextOne2, 7));
- inputEvents.add(new StreamRecord<Event>(endEvent, 8));
+ inputEvents.add(new StreamRecord<>(endEvent, 8));
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@@ -368,21 +370,21 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).followedBy("middle-first").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ }).followedByAny("middle-first").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
return value.getVolume() > 5.0;
}
- }).followedBy("middle-second").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ }).followedByAny("middle-second").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
public boolean filter(SubEvent value) throws Exception {
return value.getName().equals("next-one");
}
- }).followedBy("end").where(new SimpleCondition<Event>() {
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 7056763917392056548L;
@Override
@@ -449,28 +451,28 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
- }).oneOrMore().allowCombinations().optional().followedBy("end1").where(new SimpleCondition<Event>() {
+ }).oneOrMore().allowCombinations().optional().followedByAny("end1").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("b");
}
- }).followedBy("end2").where(new SimpleCondition<Event>() {
+ }).followedByAny("end2").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("d");
}
- }).followedBy("end3").where(new SimpleCondition<Event>() {
+ }).followedByAny("end3").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -537,7 +539,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -555,27 +557,14 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- Set<Set<Event>> resultingPatterns = new HashSet<>();
- List<Collection<Event>> allPatterns = new ArrayList<>();
-
- for (StreamRecord<Event> inputEvent : inputEvents) {
- Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getValue(),
- inputEvent.getTimestamp()).f0;
-
- for (Map<String, Event> foundPattern : patterns) {
- resultingPatterns.add(new HashSet<>(foundPattern.values()));
- allPatterns.add(foundPattern.values());
- }
- }
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
- assertEquals(4, allPatterns.size());
- assertEquals(Sets.newHashSet(
- Sets.newHashSet(startEvent, middleEvent1, middleEvent2, end1),
- Sets.newHashSet(startEvent, middleEvent1, end1),
- Sets.newHashSet(startEvent, middleEvent2, end1),
- Sets.newHashSet(startEvent, end1)
- ), resultingPatterns);
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, end1),
+ Lists.newArrayList(startEvent, middleEvent1, end1),
+ Lists.newArrayList(startEvent, middleEvent2, end1),
+ Lists.newArrayList(startEvent, end1)
+ ));
}
@Test
@@ -724,14 +713,14 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle-first").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle-first").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
- }).oneOrMore().allowCombinations().optional().followedBy("middle-second").where(new SimpleCondition<Event>() {
+ }).oneOrMore().allowCombinations().optional().followedByAny("middle-second").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -803,21 +792,21 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("branching").where(new SimpleCondition<Event>() {
+ }).followedByAny("branching").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
- }).followedBy("merging").where(new SimpleCondition<Event>() {
+ }).followedByAny("merging").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("f");
}
- }).followedBy("kleene").where(new SimpleCondition<Event>() {
+ }).followedByAny("kleene").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -940,7 +929,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("d");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -958,25 +947,12 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- Set<Set<Event>> resultingPatterns = new HashSet<>();
- List<Collection<Event>> allPatterns = new ArrayList<>();
-
- for (StreamRecord<Event> inputEvent : inputEvents) {
- Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getValue(),
- inputEvent.getTimestamp()).f0;
-
- for (Map<String, Event> foundPattern : patterns) {
- resultingPatterns.add(new HashSet<>(foundPattern.values()));
- allPatterns.add(foundPattern.values());
- }
- }
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
- assertEquals(2, allPatterns.size());
- assertEquals(Sets.newHashSet(
- Sets.newHashSet(start, middleEvent1, middleEvent2, end),
- Sets.newHashSet(start, middleEvent2, end)
- ), resultingPatterns);
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(start, middleEvent1, middleEvent2, end),
+ Lists.newArrayList(start, middleEvent2, end)
+ ));
}
@Test
@@ -1000,14 +976,14 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
- }).oneOrMore().allowCombinations().followedBy("end1").where(new SimpleCondition<Event>() {
+ }).oneOrMore().allowCombinations().followedByAny("end1").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -1183,14 +1159,14 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
- }).oneOrMore().followedBy("end1").where(new SimpleCondition<Event>() {
+ }).oneOrMore().followedByAny("end1").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -1201,26 +1177,16 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- Set<Set<Event>> resultingPatterns = new HashSet<>();
- List<Collection<Event>> allPatterns = new ArrayList<>();
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
- for (StreamRecord<Event> inputEvent : inputEvents) {
- Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getValue(),
- inputEvent.getTimestamp()).f0;
-
- for (Map<String, Event> foundPattern : patterns) {
- resultingPatterns.add(new HashSet<>(foundPattern.values()));
- allPatterns.add(foundPattern.values());
- }
- }
-
- assertEquals(3, allPatterns.size());
- assertEquals(Sets.newHashSet(
- Sets.newHashSet(startEvent, middleEvent1, middleEvent2, middleEvent3, end1),
- Sets.newHashSet(startEvent, middleEvent1, middleEvent2, end1),
- Sets.newHashSet(startEvent, middleEvent1, end1)
- ), resultingPatterns);
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3, end1),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, end1),
+ Lists.newArrayList(startEvent, middleEvent2, middleEvent3, end1),
+ Lists.newArrayList(startEvent, middleEvent3, end1),
+ Lists.newArrayList(startEvent, middleEvent2, end1),
+ Lists.newArrayList(startEvent, middleEvent1, end1)
+ ));
}
@Test
@@ -1322,25 +1288,12 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- Set<Set<Event>> resultingPatterns = new HashSet<>();
- List<Collection<Event>> allPatterns = new ArrayList<>();
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
- for (StreamRecord<Event> inputEvent : inputEvents) {
- Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getValue(),
- inputEvent.getTimestamp()).f0;
-
- for (Map<String, Event> foundPattern : patterns) {
- resultingPatterns.add(new HashSet<>(foundPattern.values()));
- allPatterns.add(foundPattern.values());
- }
- }
-
- assertEquals(2, allPatterns.size());
- assertEquals(Sets.<Set<Event>>newHashSet(
- Sets.newHashSet(startEvent, middleEvent1, middleEvent2, end1),
- Sets.newHashSet(startEvent, middleEvent1, middleEvent3, end1)
- ), resultingPatterns);
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, end1),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent3, end1)
+ ));
}
@Test
@@ -1375,25 +1328,13 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- Set<Set<Event>> resultingPatterns = new HashSet<>();
- List<Collection<Event>> allPatterns = new ArrayList<>();
-
- for (StreamRecord<Event> inputEvent : inputEvents) {
- Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getValue(),
- inputEvent.getTimestamp()).f0;
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
- for (Map<String, Event> foundPattern : patterns) {
- resultingPatterns.add(new HashSet<>(foundPattern.values()));
- allPatterns.add(foundPattern.values());
- }
- }
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(middleEvent1, middleEvent2, end1),
+ Lists.newArrayList(middleEvent2, middleEvent3, end1)
+ ));
- assertEquals(2, allPatterns.size());
- assertEquals(Sets.<Set<Event>>newHashSet(
- Sets.newHashSet(middleEvent1, middleEvent2, end1),
- Sets.newHashSet(middleEvent2, middleEvent3, end1)
- ), resultingPatterns);
}
@Test
@@ -1442,49 +1383,6 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesStrictWithFollowedBy() {
- List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
- inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent1, 2));
- inputEvents.add(new StreamRecord<>(new Event(23, "f", 1.0), 3));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent2, 4));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 6));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.end, 7));
-
- Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("c");
- }
- }).followedBy("middle").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("a");
- }
- }).times(2).allowCombinations().consecutive().followedBy("end1").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("b");
- }
- });
-
- 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.startEvent, ConsecutiveData.middleEvent3, ConsecutiveData.middleEvent2, ConsecutiveData.end)
- ));
- }
-
- @Test
public void testTimesNotStrictWithFollowedByEager() {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
@@ -1543,7 +1441,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -1571,49 +1469,6 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testTimesStrictWithFollowedByNotEager() {
- List<StreamRecord<Event>> inputEvents = new ArrayList<>();
-
- inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent1, 2));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent2, 4));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 6));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.end, 7));
-
- Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("c");
- }
- }).followedBy("middle").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("a");
- }
- }).times(2).allowCombinations().consecutive().followedBy("end1").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("b");
- }
- });
-
- 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.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end)
- ));
- }
-
- @Test
public void testTimesStrictWithNextAndConsecutive() {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
@@ -1789,29 +1644,16 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- Set<Set<Event>> resultingPatterns = new HashSet<>();
- List<Collection<Event>> allPatterns = new ArrayList<>();
-
- for (StreamRecord<Event> inputEvent : inputEvents) {
- Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getValue(),
- inputEvent.getTimestamp()).f0;
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
- for (Map<String, Event> foundPattern : patterns) {
- resultingPatterns.add(new HashSet<>(foundPattern.values()));
- allPatterns.add(foundPattern.values());
- }
- }
-
- assertEquals(6, allPatterns.size());
- assertEquals(Sets.newHashSet(
- Sets.newHashSet(middleEvent1, middleEvent2, middleEvent3),
- Sets.newHashSet(middleEvent1, middleEvent2),
- Sets.newHashSet(middleEvent1),
- Sets.newHashSet(middleEvent2, middleEvent3),
- Sets.newHashSet(middleEvent2),
- Sets.newHashSet(middleEvent3)
- ), resultingPatterns);
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(middleEvent1, middleEvent2, middleEvent3),
+ Lists.newArrayList(middleEvent1, middleEvent2),
+ Lists.newArrayList(middleEvent1),
+ Lists.newArrayList(middleEvent2, middleEvent3),
+ Lists.newArrayList(middleEvent2),
+ Lists.newArrayList(middleEvent3)
+ ));
}
@Test
@@ -1895,26 +1737,13 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- Set<Set<Event>> resultingPatterns = new HashSet<>();
- List<Collection<Event>> allPatterns = new ArrayList<>();
-
- for (StreamRecord<Event> inputEvent : inputEvents) {
- Collection<Map<String, Event>> patterns = nfa.process(
- inputEvent.getValue(),
- inputEvent.getTimestamp()).f0;
-
- for (Map<String, Event> foundPattern : patterns) {
- resultingPatterns.add(new HashSet<>(foundPattern.values()));
- allPatterns.add(foundPattern.values());
- }
- }
+ final List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
- assertEquals(3, allPatterns.size());
- assertEquals(Sets.newHashSet(
- Sets.newHashSet(startEvent, middleEvent1, middleEvent2, middleEvent3),
- Sets.newHashSet(startEvent, middleEvent1, middleEvent2),
- Sets.newHashSet(startEvent, middleEvent1)
- ), resultingPatterns);
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2, middleEvent3),
+ Lists.newArrayList(startEvent, middleEvent1, middleEvent2),
+ Lists.newArrayList(startEvent, middleEvent1)
+ ));
}
/////////////////////////////// Optional ////////////////////////////////////////
@@ -1978,7 +1807,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -2025,7 +1854,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -2071,7 +1900,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -2098,6 +1927,54 @@ public class NFAITCase extends TestLogger {
}
@Test
+ public void testOneOrMoreStrictOptional() {
+ List<StreamRecord<Event>> inputEvents = new ArrayList<>();
+
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
+ inputEvents.add(new StreamRecord<>(new Event(23, "f", 1.0), 2));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent1, 3));
+ inputEvents.add(new StreamRecord<>(new Event(23, "f", 1.0), 4));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent2, 5));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 6));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.end, 7));
+
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("c");
+ }
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("a");
+ }
+ }).oneOrMore().consecutive().optional().followedBy("end1").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("b");
+ }
+ });
+
+ 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.startEvent, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent2, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent3, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)
+ ));
+ }
+
+ @Test
public void testTimesStrictOptional1() {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
@@ -2201,94 +2078,92 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictCombinationsOneOrMore() {
- List<List<Event>> resultingPatterns = testStrictOneOrMore(false);
+ public void testStrictOneOrMore() {
+ List<List<Event>> resultingPatterns = testOneOrMore(Quantifier.ConsumingStrategy.STRICT);
compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent2, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent3, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent4, ConsecutiveData.end)
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end)
+ ));
+ }
+
+ @Test
+ public void testSkipTillNextOneOrMore() {
+ List<List<Event>> resultingPatterns = testOneOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_NEXT);
+
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.middleEvent4, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end)
));
}
@Test
- public void testStrictEagerOneOrMore() {
- List<List<Event>> resultingPatterns = testStrictOneOrMore(true);
+ public void testSkipTillAnyOneOrMore() {
+ List<List<Event>> resultingPatterns = testOneOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_ANY);
compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.middleEvent4, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent3, ConsecutiveData.middleEvent4, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent4, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent3, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent4, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end)
));
}
- private List<List<Event>> testStrictOneOrMore(boolean eager) {
+ private List<List<Event>> testOneOrMore(Quantifier.ConsumingStrategy strategy) {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
inputEvents.add(new StreamRecord<>(new Event(50, "d", 6.0), 2));
inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent1, 3));
inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent2, 4));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 5));
- inputEvents.add(new StreamRecord<>(new Event(50, "d", 6.0), 6));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent4, 7));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.end, 8));
-
- Pattern<Event, ?> pattern;
- if (eager) {
- pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("c");
- }
- }).followedBy("middle").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("a");
- }
- }).oneOrMore().consecutive()
- .followedBy("end1").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 4));
+ inputEvents.add(new StreamRecord<>(new Event(50, "d", 6.0), 5));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent4, 6));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.end, 7));
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("b");
- }
- });
- } else {
- pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("c");
- }
- }).followedBy("middle").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("c");
+ }
+ }).followedBy("middle").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("a");
- }
- }).oneOrMore().allowCombinations().consecutive()
- .followedBy("end1").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("a");
+ }
+ }).oneOrMore();
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("b");
- }
- });
+ switch (strategy) {
+ case STRICT:
+ pattern = pattern.consecutive();
+ break;
+ case SKIP_TILL_NEXT:
+ break;
+ case SKIP_TILL_ANY:
+ pattern = pattern.allowCombinations();
+ break;
}
+ pattern = pattern.followedBy("end1").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("b");
+ }
+ });
+
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
return feedNFA(inputEvents, nfa);
@@ -2296,9 +2171,10 @@ public class NFAITCase extends TestLogger {
@Test
public void testStrictEagerZeroOrMore() {
- List<List<Event>> resultingPatterns = testStrictZeroOrMore(true);
+ List<List<Event>> resultingPatterns = testZeroOrMore(Quantifier.ConsumingStrategy.STRICT);
compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)
@@ -2306,67 +2182,75 @@ public class NFAITCase extends TestLogger {
}
@Test
- public void testStrictCombinationsZeroOrMore() {
- List<List<Event>> resultingPatterns = testStrictZeroOrMore(false);
+ public void testSkipTillAnyZeroOrMore() {
+ List<List<Event>> resultingPatterns = testZeroOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_ANY);
compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.middleEvent4, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent4, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent3, ConsecutiveData.middleEvent4, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent4, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent3, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent2, ConsecutiveData.end),
- Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent3, ConsecutiveData.end),
Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)
));
}
- private List<List<Event>> testStrictZeroOrMore(boolean eager) {
+ @Test
+ public void testSkipTillNextZeroOrMore() {
+ List<List<Event>> resultingPatterns = testZeroOrMore(Quantifier.ConsumingStrategy.SKIP_TILL_NEXT);
+
+ compareMaps(resultingPatterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.middleEvent4, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.middleEvent3, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.middleEvent2, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.middleEvent1, ConsecutiveData.end),
+ Lists.newArrayList(ConsecutiveData.startEvent, ConsecutiveData.end)
+ ));
+ }
+
+ private List<List<Event>> testZeroOrMore(Quantifier.ConsumingStrategy strategy) {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
inputEvents.add(new StreamRecord<>(ConsecutiveData.startEvent, 1));
inputEvents.add(new StreamRecord<>(new Event(50, "d", 6.0), 2));
inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent1, 3));
inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent2, 4));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 4));
inputEvents.add(new StreamRecord<>(new Event(50, "d", 6.0), 5));
- inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent3, 6));
+ inputEvents.add(new StreamRecord<>(ConsecutiveData.middleEvent4, 6));
inputEvents.add(new StreamRecord<>(ConsecutiveData.end, 7));
- Pattern<Event, ?> pattern = eager
- ? Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
-
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("c");
- }
- }).followedBy("middle").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("a");
- }
- }).oneOrMore().optional().consecutive().followedBy("end1").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("c");
+ }
+ }).followedBy("middle").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("b");
- }
- })
- : Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("a");
+ }
+ }).oneOrMore().optional();
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("c");
- }
- }).followedBy("middle").where(new SimpleCondition<Event>() {
- private static final long serialVersionUID = 5726188262756267490L;
+ switch (strategy) {
+ case STRICT:
+ pattern = pattern.consecutive();
+ break;
+ case SKIP_TILL_NEXT:
+ break;
+ case SKIP_TILL_ANY:
+ pattern = pattern.allowCombinations();
+ break;
+ }
- @Override
- public boolean filter(Event value) throws Exception {
- return value.getName().equals("a");
- }
- }).oneOrMore().allowCombinations().optional().consecutive().followedBy("end1").where(new SimpleCondition<Event>() {
+ pattern = pattern.followedBy("end1").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -2380,7 +2264,6 @@ public class NFAITCase extends TestLogger {
return feedNFA(inputEvents, nfa);
}
-
@Test
public void testTimesStrict() {
List<StreamRecord<Event>> inputEvents = new ArrayList<>();
@@ -2400,14 +2283,14 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("a");
}
- }).times(2).allowCombinations().consecutive().followedBy("end1").where(new SimpleCondition<Event>() {
+ }).times(2).consecutive().followedBy("end1").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -2444,7 +2327,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("c");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 5726188262756267490L;
@Override
@@ -2733,16 +2616,10 @@ public class NFAITCase extends TestLogger {
Lists.<List<Event>>newArrayList(
Lists.newArrayList(startEvent1, endEvent, middleEvent1, middleEvent2, middleEvent4),
Lists.newArrayList(startEvent1, endEvent, middleEvent2, middleEvent1),
- Lists.newArrayList(startEvent1, endEvent, middleEvent4, middleEvent3),
- Lists.newArrayList(startEvent1, endEvent, middleEvent4, middleEvent2),
Lists.newArrayList(startEvent1, endEvent, middleEvent3, middleEvent1),
Lists.newArrayList(startEvent2, endEvent, middleEvent3, middleEvent4),
Lists.newArrayList(startEvent1, endEvent, middleEvent4, middleEvent1),
- Lists.newArrayList(startEvent1, endEvent, middleEvent4),
Lists.newArrayList(startEvent1, endEvent, middleEvent1),
- Lists.newArrayList(startEvent1, endEvent, middleEvent2),
- Lists.newArrayList(startEvent1, endEvent, middleEvent3),
- Lists.newArrayList(startEvent2, endEvent, middleEvent4),
Lists.newArrayList(startEvent2, endEvent, middleEvent3)
)
);
@@ -2798,9 +2675,7 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
- return resultingPatterns;
+ return feedNFA(inputEvents, nfa);
}
private static class MySubeventIterCondition extends IterativeCondition<SubEvent> {
@@ -2887,9 +2762,7 @@ public class NFAITCase extends TestLogger {
NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
- List<List<Event>> resultingPatterns = feedNFA(inputEvents, nfa);
-
- return resultingPatterns;
+ return feedNFA(inputEvents, nfa);
}
private static class MyEventIterCondition extends IterativeCondition<Event> {
@@ -2978,7 +2851,7 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).followedBy("middle1").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ }).followedByAny("middle1").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
private static final long serialVersionUID = 2178338526904474690L;
@Override
@@ -3043,14 +2916,14 @@ public class NFAITCase extends TestLogger {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).oneOrMore().followedBy("middle1").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ }).oneOrMore().followedByAny("middle1").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
private static final long serialVersionUID = 2178338526904474690L;
@Override
public boolean filter(SubEvent value) throws Exception {
return value.getName().startsWith("foo");
}
- }).followedBy("end").where(new IterativeCondition<Event>() {
+ }).followedByAny("end").where(new IterativeCondition<Event>() {
private static final long serialVersionUID = 7056763917392056548L;
@Override
@@ -3084,6 +2957,131 @@ public class NFAITCase extends TestLogger {
);
}
+
+ /////////////////////////////////////// Skip till next /////////////////////////////
+
+ @Test
+ public void testBranchingPatternSkipTillNext() {
+ List<StreamRecord<Event>> inputEvents = new ArrayList<>();
+
+ Event startEvent = new Event(40, "start", 1.0);
+ SubEvent middleEvent1 = new SubEvent(41, "foo1", 1.0, 10.0);
+ SubEvent middleEvent2 = new SubEvent(42, "foo2", 1.0, 10.0);
+ SubEvent middleEvent3 = new SubEvent(43, "foo3", 1.0, 10.0);
+ SubEvent nextOne1 = new SubEvent(44, "next-one", 1.0, 2.0);
+ SubEvent nextOne2 = new SubEvent(45, "next-one", 1.0, 2.0);
+ Event endEvent= new Event(46, "end", 1.0);
+
+ inputEvents.add(new StreamRecord<>(startEvent, 1));
+ inputEvents.add(new StreamRecord<Event>(middleEvent1, 3));
+ inputEvents.add(new StreamRecord<Event>(middleEvent2, 4));
+ inputEvents.add(new StreamRecord<Event>(middleEvent3, 5));
+ inputEvents.add(new StreamRecord<Event>(nextOne1, 6));
+ inputEvents.add(new StreamRecord<Event>(nextOne2, 7));
+ inputEvents.add(new StreamRecord<>(endEvent, 8));
+
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("start");
+ }
+ }).followedBy("middle-first").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ private static final long serialVersionUID = 6215754202506583964L;
+
+ @Override
+ public boolean filter(SubEvent value) throws Exception {
+ return value.getVolume() > 5.0;
+ }
+ }).followedBy("middle-second").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ private static final long serialVersionUID = 6215754202506583964L;
+
+ @Override
+ public boolean filter(SubEvent value) throws Exception {
+ return value.getName().equals("next-one");
+ }
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 7056763917392056548L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("end");
+ }
+ });
+
+ NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
+
+ final List<List<Event>> patterns = feedNFA(inputEvents, nfa);
+
+ compareMaps(patterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(startEvent, middleEvent1, nextOne1, endEvent)
+ ));
+ }
+
+ @Test
+ public void testBranchingPatternMixedFollowedBy() {
+ List<StreamRecord<Event>> inputEvents = new ArrayList<>();
+
+ Event startEvent = new Event(40, "start", 1.0);
+ SubEvent middleEvent1 = new SubEvent(41, "foo1", 1.0, 10.0);
+ SubEvent middleEvent2 = new SubEvent(42, "foo2", 1.0, 10.0);
+ SubEvent middleEvent3 = new SubEvent(43, "foo3", 1.0, 10.0);
+ SubEvent nextOne1 = new SubEvent(44, "next-one", 1.0, 2.0);
+ SubEvent nextOne2 = new SubEvent(45, "next-one", 1.0, 2.0);
+ Event endEvent= new Event(46, "end", 1.0);
+
+ inputEvents.add(new StreamRecord<>(startEvent, 1));
+ inputEvents.add(new StreamRecord<Event>(middleEvent1, 3));
+ inputEvents.add(new StreamRecord<Event>(middleEvent2, 4));
+ inputEvents.add(new StreamRecord<Event>(middleEvent3, 5));
+ inputEvents.add(new StreamRecord<Event>(nextOne1, 6));
+ inputEvents.add(new StreamRecord<Event>(nextOne2, 7));
+ inputEvents.add(new StreamRecord<>(endEvent, 8));
+
+ Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 5726188262756267490L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("start");
+ }
+ }).followedByAny("middle-first").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ private static final long serialVersionUID = 6215754202506583964L;
+
+ @Override
+ public boolean filter(SubEvent value) throws Exception {
+ return value.getVolume() > 5.0;
+ }
+ }).followedBy("middle-second").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ private static final long serialVersionUID = 6215754202506583964L;
+
+ @Override
+ public boolean filter(SubEvent value) throws Exception {
+ return value.getName().equals("next-one");
+ }
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
+ private static final long serialVersionUID = 7056763917392056548L;
+
+ @Override
+ public boolean filter(Event value) throws Exception {
+ return value.getName().equals("end");
+ }
+ });
+
+ NFA<Event> nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);
+
+ final List<List<Event>> patterns = feedNFA(inputEvents, nfa);
+
+ compareMaps(patterns, Lists.<List<Event>>newArrayList(
+ Lists.newArrayList(startEvent, middleEvent1, nextOne1, endEvent),
+ Lists.newArrayList(startEvent, middleEvent2, nextOne1, endEvent),
+ Lists.newArrayList(startEvent, middleEvent3, nextOne1, endEvent)
+ ));
+ }
+
+
+ ///////////////////////////////////////// Utility ////////////////////////////////
private List<List<Event>> feedNFA(List<StreamRecord<Event>> inputEvents, NFA<Event> nfa) {
List<List<Event>> resultingPatterns = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java
index da1f9c8..ced9efe 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/compiler/NFACompilerTest.java
@@ -149,45 +149,6 @@ public class NFACompilerTest extends TestLogger {
assertEquals(0, endingState.getStateTransitions().size());
}
- @Test
- public void testNFACompilerWithKleeneStar() {
-
- Pattern<Event, Event> pattern = Pattern.<Event>begin("start").where(startFilter)
- .followedBy("middle").subtype(SubEvent.class).oneOrMore().optional()
- .followedBy("end").where(endFilter);
-
- NFA<Event> nfa = NFACompiler.compile(pattern, serializer, false);
-
- Set<State<Event>> states = nfa.getStates();
- assertEquals(5, states.size());
-
-
- Set<Tuple2<String, Set<Tuple2<String, StateTransitionAction>>>> stateMap = new HashSet<>();
- for (State<Event> state : states) {
- stateMap.add(Tuple2.of(state.getName(), unfoldTransitions(state)));
- }
-
- assertEquals(stateMap, newHashSet(
- Tuple2.of("start", newHashSet(Tuple2.of("middle", StateTransitionAction.TAKE))),
- Tuple2.of("middle", newHashSet(
- Tuple2.of("middle", StateTransitionAction.IGNORE),
- Tuple2.of("middle", StateTransitionAction.TAKE)
- )),
- Tuple2.of("middle", newHashSet(
- Tuple2.of("middle", StateTransitionAction.IGNORE),
- Tuple2.of("middle", StateTransitionAction.TAKE),
- Tuple2.of("end", StateTransitionAction.PROCEED)
- )),
- Tuple2.of("end", newHashSet(
- Tuple2.of(NFACompiler.ENDING_STATE_NAME, StateTransitionAction.TAKE),
- Tuple2.of("end", StateTransitionAction.IGNORE)
- )),
- Tuple2.of(NFACompiler.ENDING_STATE_NAME, Sets.newHashSet())
- ));
-
- }
-
-
private <T> Set<Tuple2<String, StateTransitionAction>> unfoldTransitions(final State<T> state) {
final Set<Tuple2<String, StateTransitionAction>> transitions = new HashSet<>();
for (StateTransition<T> transition : state.getStateTransitions()) {
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
index d599ec9..8465bc3 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
@@ -208,6 +208,7 @@ public class CEPOperatorTest extends TestLogger {
* Tests that the internal time of a CEP operator advances only given watermarks. See FLINK-5033
*/
@Test
+ @SuppressWarnings("unchecked")
public void testKeyedAdvancingTimeWithoutElements() throws Exception {
final KeySelector<Event, Integer> keySelector = new TestKeySelector();
@@ -522,7 +523,7 @@ public class CEPOperatorTest extends TestLogger {
return value.getName().equals("start");
}
})
- .followedBy("middle").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
+ .followedByAny("middle").subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
private static final long serialVersionUID = 6215754202506583964L;
@Override
@@ -530,7 +531,7 @@ public class CEPOperatorTest extends TestLogger {
return value.getVolume() > 5.0;
}
})
- .followedBy("end").where(new SimpleCondition<Event>() {
+ .followedByAny("end").where(new SimpleCondition<Event>() {
private static final long serialVersionUID = 7056763917392056548L;
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java
index 8c4304a..3c1da2e 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/pattern/PatternTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.cep.pattern;
import org.apache.flink.cep.Event;
import org.apache.flink.cep.SubEvent;
+import org.apache.flink.cep.pattern.Quantifier.ConsumingStrategy;
import org.apache.flink.cep.pattern.conditions.OrCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.cep.pattern.conditions.SubtypeCondition;
@@ -58,8 +59,8 @@ public class PatternTest extends TestLogger {
assertNotNull(previous2 = previous.getPrevious());
assertNull(previous2.getPrevious());
- assertTrue(pattern instanceof FollowedByPattern);
- assertTrue(previous instanceof FollowedByPattern);
+ assertEquals(ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier().getConsumingStrategy());
+ assertEquals(ConsumingStrategy.SKIP_TILL_NEXT, previous.getQuantifier().getConsumingStrategy());
assertEquals(pattern.getName(), "end");
assertEquals(previous.getName(), "next");
@@ -137,7 +138,7 @@ public class PatternTest extends TestLogger {
assertNotNull(previous2 = previous.getPrevious());
assertNull(previous2.getPrevious());
- assertTrue(pattern instanceof FollowedByPattern);
+ assertEquals(ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier().getConsumingStrategy());
assertNotNull(previous.getCondition());
assertEquals(pattern.getName(), "end");
@@ -177,7 +178,7 @@ public class PatternTest extends TestLogger {
assertNotNull(previous2 = previous.getPrevious());
assertNull(previous2.getPrevious());
- assertTrue(pattern instanceof FollowedByPattern);
+ assertEquals(ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier().getConsumingStrategy());
assertFalse(previous.getCondition() instanceof OrCondition);
assertTrue(previous2.getCondition() instanceof OrCondition);
[2/2] flink git commit: [FLINK-6208] [cep] Implement skip till next
match strategy
Posted by kk...@apache.org.
[FLINK-6208] [cep] Implement skip till next match strategy
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7c35dc0e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7c35dc0e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7c35dc0e
Branch: refs/heads/master
Commit: 7c35dc0e40a96f6b8d367b9a80a19e9a22525043
Parents: b527582
Author: Dawid Wysakowicz <da...@getindata.com>
Authored: Wed Apr 26 11:57:53 2017 +0200
Committer: kl0u <kk...@gmail.com>
Committed: Fri Apr 28 14:28:37 2017 +0200
----------------------------------------------------------------------
docs/dev/libs/cep.md | 19 +-
.../cep/scala/pattern/FollowedByPattern.scala | 44 --
.../flink/cep/scala/pattern/Pattern.scala | 31 +-
.../flink/cep/scala/pattern/package.scala | 3 +-
.../flink/cep/scala/pattern/PatternTest.scala | 52 +-
.../flink/cep/nfa/compiler/NFACompiler.java | 192 ++---
.../flink/cep/pattern/FollowedByPattern.java | 33 -
.../org/apache/flink/cep/pattern/Pattern.java | 42 +-
.../apache/flink/cep/pattern/Quantifier.java | 82 +-
.../java/org/apache/flink/cep/CEPITCase.java | 30 +-
.../org/apache/flink/cep/nfa/NFAITCase.java | 760 +++++++++----------
.../flink/cep/nfa/compiler/NFACompilerTest.java | 39 -
.../flink/cep/operator/CEPOperatorTest.java | 5 +-
.../apache/flink/cep/pattern/PatternTest.java | 9 +-
14 files changed, 656 insertions(+), 685 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/docs/dev/libs/cep.md
----------------------------------------------------------------------
diff --git a/docs/dev/libs/cep.md b/docs/dev/libs/cep.md
index 15afdf5..b379615 100644
--- a/docs/dev/libs/cep.md
+++ b/docs/dev/libs/cep.md
@@ -268,7 +268,7 @@ val strictNext: Pattern[Event, _] = start.next("middle")
</div>
Non-strict contiguity means that other events are allowed to occur in-between two matching events.
-A non-strict contiguity pattern state can be created via the `followedBy` method.
+A non-strict contiguity pattern state can be created via the `followedBy` or `followedByAny` method.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
@@ -282,6 +282,23 @@ Pattern<Event, ?> nonStrictNext = start.followedBy("middle");
val nonStrictNext : Pattern[Event, _] = start.followedBy("middle")
{% endhighlight %}
</div>
+
+For non-strict contiguity one can specify if only the first succeeding matching event will be matched, or
+all. In the latter case multiple matches will be emitted for the same beginning.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+Pattern<Event, ?> nonStrictNext = start.followedByAny("middle");
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+val nonStrictNext : Pattern[Event, _] = start.followedByAny("middle")
+{% endhighlight %}
+</div>
+
</div>
It is also possible to define a temporal constraint for the pattern to be valid.
For example, one can define that a pattern should occur within 10 seconds via the `within` method.
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/FollowedByPattern.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/FollowedByPattern.scala b/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/FollowedByPattern.scala
deleted file mode 100644
index 4bda08f..0000000
--- a/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/FollowedByPattern.scala
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.cep.scala.pattern
-
-import org.apache.flink.cep.pattern.{FollowedByPattern => JFollowedByPattern}
-
-object FollowedByPattern {
- /**
- * Constructs a new Pattern by wrapping a given Java API Pattern
- *
- * @param jfbPattern Underlying Java API Pattern.
- * @tparam T Base type of the elements appearing in the pattern
- * @tparam F Subtype of T to which the current pattern operator is constrained
- * @return New wrapping FollowedByPattern object
- */
- def apply[T, F <: T](jfbPattern: JFollowedByPattern[T, F]) =
- new FollowedByPattern[T, F](jfbPattern)
-}
-
-/**
- * Pattern operator which signifies that the there is a non-strict temporal contiguity between
- * itself and its preceding pattern operator. This means that there might be events in between
- * two matching events. These events are then simply ignored.
- *
- * @tparam T Base type of the events
- * @tparam F Subtype of T to which the operator is currently constrained
- */
-class FollowedByPattern[T, F <: T](jfbPattern: JFollowedByPattern[T, F])
- extends Pattern[T, F](jfbPattern)
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/Pattern.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/Pattern.scala b/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/Pattern.scala
index 3bdbfcf..65b7ab0 100644
--- a/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/Pattern.scala
+++ b/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/Pattern.scala
@@ -46,22 +46,22 @@ class Pattern[T , F <: T](jPattern: JPattern[T, F]) {
/**
* @return The previous pattern
*/
- def getPrevious(): Option[Pattern[T, _ <: T]] = {
- wrapPattern(jPattern.getPrevious())
+ def getPrevious: Option[Pattern[T, _ <: T]] = {
+ wrapPattern(jPattern.getPrevious)
}
/**
*
* @return Name of the pattern operator
*/
- def getName(): String = jPattern.getName()
+ def getName: String = jPattern.getName
/**
*
* @return Window length in which the pattern match has to occur
*/
- def getWindowTime(): Option[Time] = {
- Option(jPattern.getWindowTime())
+ def getWindowTime: Option[Time] = {
+ Option(jPattern.getWindowTime)
}
/**
@@ -70,8 +70,8 @@ class Pattern[T , F <: T](jPattern: JPattern[T, F]) {
*/
def getQuantifier: Quantifier = jPattern.getQuantifier
- def getCondition(): Option[IterativeCondition[F]] = {
- Option(jPattern.getCondition())
+ def getCondition: Option[IterativeCondition[F]] = {
+ Option(jPattern.getCondition)
}
/**
@@ -208,7 +208,22 @@ class Pattern[T , F <: T](jPattern: JPattern[T, F]) {
* @param name Name of the new pattern
* @return A new pattern which is appended to this one
*/
- def followedBy(name: String) = FollowedByPattern(jPattern.followedBy(name))
+ def followedBy(name: String): Pattern[T, T] = {
+ Pattern[T, T](jPattern.followedBy(name))
+ }
+
+ /**
+ * Appends a new pattern to the existing one. The new pattern enforces non-strict
+ * temporal contiguity. This means that a matching event of this pattern and the
+ * preceding matching event might be interleaved with other events which are ignored.
+ *
+ * @param name Name of the new pattern
+ * @return A new pattern which is appended to this one
+ */
+ def followedByAny(name: String): Pattern[T, T] = {
+ Pattern[T, T](jPattern.followedByAny(name))
+ }
+
/**
* Specifies that this pattern is optional for a final match of the pattern
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/package.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/package.scala b/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/package.scala
index 382c160..26355a5 100644
--- a/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/package.scala
+++ b/flink-libraries/flink-cep-scala/src/main/scala/org/apache/flink/cep/scala/pattern/package.scala
@@ -17,7 +17,7 @@
*/
package org.apache.flink.cep.scala
-import org.apache.flink.cep.pattern.{FollowedByPattern => JFollowedByPattern, Pattern => JPattern}
+import org.apache.flink.cep.pattern.{Pattern => JPattern}
package object pattern {
/**
@@ -31,7 +31,6 @@ package object pattern {
*/
private[flink] def wrapPattern[T, F <: T](javaPattern: JPattern[T, F])
: Option[Pattern[T, F]] = javaPattern match {
- case f: JFollowedByPattern[T, F] => Some(FollowedByPattern[T, F](f))
case p: JPattern[T, F] => Some(Pattern[T, F](p))
case _ => None
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep-scala/src/test/scala/org/apache/flink/cep/scala/pattern/PatternTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep-scala/src/test/scala/org/apache/flink/cep/scala/pattern/PatternTest.scala b/flink-libraries/flink-cep-scala/src/test/scala/org/apache/flink/cep/scala/pattern/PatternTest.scala
index a95dddd..d574513 100644
--- a/flink-libraries/flink-cep-scala/src/test/scala/org/apache/flink/cep/scala/pattern/PatternTest.scala
+++ b/flink-libraries/flink-cep-scala/src/test/scala/org/apache/flink/cep/scala/pattern/PatternTest.scala
@@ -22,7 +22,7 @@ import org.junit.Assert._
import org.junit.Test
import org.apache.flink.cep.Event
import org.apache.flink.cep.SubEvent
-import org.apache.flink.cep.pattern.conditions.IterativeCondition.Context
+import org.apache.flink.cep.pattern.Quantifier.ConsumingStrategy
import org.apache.flink.cep.pattern.conditions._
class PatternTest {
@@ -33,7 +33,7 @@ class PatternTest {
*/
@Test
- def testStrictContiguity: Unit = {
+ def testStrictContiguity(): Unit = {
val pattern = Pattern.begin[Event]("start").next("next").next("end")
val jPattern = JPattern.begin[Event]("start").next("next").next("end")
@@ -56,7 +56,7 @@ class PatternTest {
@Test
- def testNonStrictContiguity: Unit = {
+ def testNonStrictContiguity(): Unit = {
val pattern = Pattern.begin[Event]("start").followedBy("next").followedBy("end")
val jPattern = JPattern.begin[Event]("start").followedBy("next").followedBy("end")
@@ -69,8 +69,8 @@ class PatternTest {
assertTrue(previous.getPrevious.isDefined)
assertFalse(preprevious.getPrevious.isDefined)
- assertTrue(pattern.isInstanceOf[FollowedByPattern[_, _]])
- assertTrue(previous.isInstanceOf[FollowedByPattern[_, _]])
+ assertEquals(ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier.getConsumingStrategy)
+ assertEquals(ConsumingStrategy.SKIP_TILL_NEXT, previous.getQuantifier.getConsumingStrategy)
assertEquals(pattern.getName, "end")
assertEquals(previous.getName, "next")
@@ -78,25 +78,25 @@ class PatternTest {
}
@Test
- def testStrictContiguityWithCondition: Unit = {
+ def testStrictContiguityWithCondition(): Unit = {
val pattern = Pattern.begin[Event]("start")
.next("next")
- .where((value: Event, ctx: Context[Event]) => value.getName() == "foobar")
+ .where((value: Event, _) => value.getName == "foobar")
.next("end")
- .where((value: Event, ctx: Context[Event]) => value.getId() == 42)
+ .where((value: Event, _) => value.getId == 42)
val jPattern = JPattern.begin[Event]("start")
.next("next")
.where(new SimpleCondition[Event]() {
@throws[Exception]
def filter(value: Event): Boolean = {
- return value.getName() == "foobar"
+ value.getName == "foobar"
}
}).next("end")
.where(new SimpleCondition[Event]() {
@throws[Exception]
def filter(value: Event): Boolean = {
- return value.getId() == 42
+ value.getId == 42
}
})
@@ -120,7 +120,7 @@ class PatternTest {
}
@Test
- def testPatternWithSubtyping: Unit = {
+ def testPatternWithSubtyping(): Unit = {
val pattern = Pattern.begin[Event]("start")
.next("subevent")
.subtype(classOf[SubEvent])
@@ -150,11 +150,11 @@ class PatternTest {
}
@Test
- def testPatternWithSubtypingAndFilter: Unit = {
+ def testPatternWithSubtypingAndFilter(): Unit = {
val pattern = Pattern.begin[Event]("start")
.next("subevent")
.subtype(classOf[SubEvent])
- .where((value: SubEvent) => false)
+ .where(_ => false)
.followedBy("end")
val jpattern = JPattern.begin[Event]("start")
@@ -163,7 +163,7 @@ class PatternTest {
.where(new SimpleCondition[SubEvent]() {
@throws[Exception]
def filter(value: SubEvent): Boolean = {
- return false
+ false
}
}).followedBy("end")
@@ -178,8 +178,8 @@ class PatternTest {
assertTrue(previous.getPrevious.isDefined)
assertFalse(preprevious.getPrevious.isDefined)
- assertTrue(pattern.isInstanceOf[FollowedByPattern[_, _]])
- assertTrue(previous.getCondition().isDefined)
+ assertEquals(ConsumingStrategy.SKIP_TILL_NEXT, pattern.getQuantifier.getConsumingStrategy)
+ assertTrue(previous.getCondition.isDefined)
assertEquals(pattern.getName, "end")
assertEquals(previous.getName, "subevent")
@@ -194,25 +194,25 @@ class PatternTest {
&& threeWayEquals(
pattern.getName,
pattern.wrappedPattern.getName,
- jPattern.getName())
+ jPattern.getName)
//check equal time windows
&& threeWayEquals(
pattern.getWindowTime.orNull,
pattern.wrappedPattern.getWindowTime,
- jPattern.getWindowTime())
+ jPattern.getWindowTime)
//check congruent class names / types
&& threeWayEquals(
pattern.getClass.getSimpleName,
pattern.wrappedPattern.getClass.getSimpleName,
- jPattern.getClass().getSimpleName())
+ jPattern.getClass.getSimpleName)
//best effort to confirm congruent filter functions
&& compareFilterFunctions(
- pattern.getCondition().orNull,
- jPattern.getCondition())
+ pattern.getCondition.orNull,
+ jPattern.getCondition)
//recursively check previous patterns
&& checkCongruentRepresentations(
pattern.getPrevious.orNull,
- jPattern.getPrevious()))
+ jPattern.getPrevious))
}
def threeWayEquals(a: AnyRef, b: AnyRef, c: AnyRef): Boolean = {
@@ -233,15 +233,15 @@ class PatternTest {
(sFilter, jFilter) match {
//matching types: and-filter; branch and recurse for inner filters
case (saf: AndCondition[_], jaf: AndCondition[_])
- => (compareFilterFunctions(saf.getLeft(), jaf.getLeft())
- && compareFilterFunctions(saf.getRight(), jaf.getRight()))
+ => (compareFilterFunctions(saf.getLeft, jaf.getLeft)
+ && compareFilterFunctions(saf.getRight, jaf.getRight))
//matching types: subtype-filter
- case (saf: SubtypeCondition[_], jaf: SubtypeCondition[_]) => true
+ case (_: SubtypeCondition[_], _: SubtypeCondition[_]) => true
//mismatch: one-sided and/subtype-filter
case (_: AndCondition[_] | _: SubtypeCondition[_], _) => false
case (_, _: AndCondition[_] | _: SubtypeCondition[_]) => false
//from here we can only check mutual presence or absence of a function
- case (s: IterativeCondition[_], j: IterativeCondition[_]) => true
+ case (_: IterativeCondition[_], _: IterativeCondition[_]) => true
case (null, null) => true
case _ => false
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java
index b4e0557..0ca0e14 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java
@@ -20,32 +20,30 @@ package org.apache.flink.cep.nfa.compiler;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterators;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.State;
import org.apache.flink.cep.nfa.StateTransition;
import org.apache.flink.cep.nfa.StateTransitionAction;
-import org.apache.flink.cep.pattern.conditions.BooleanConditions;
-import org.apache.flink.cep.pattern.FollowedByPattern;
import org.apache.flink.cep.pattern.MalformedPatternException;
-import org.apache.flink.cep.pattern.conditions.NotCondition;
import org.apache.flink.cep.pattern.Pattern;
-import org.apache.flink.cep.pattern.Quantifier.QuantifierProperty;
+import org.apache.flink.cep.pattern.Quantifier;
+import org.apache.flink.cep.pattern.conditions.BooleanConditions;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
+import org.apache.flink.cep.pattern.conditions.NotCondition;
import org.apache.flink.streaming.api.windowing.time.Time;
-import javax.annotation.Nullable;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
/**
* Compiler class containing methods to compile a {@link Pattern} into a {@link NFA} or a
* {@link NFAFactory}.
@@ -159,25 +157,7 @@ public class NFACompiler {
State<T> lastSink = sinkState;
while (currentPattern.getPrevious() != null) {
- checkPatternNameUniqueness();
- usedNames.add(currentPattern.getName());
-
- if (currentPattern.getQuantifier().hasProperty(QuantifierProperty.LOOPING)) {
- final State<T> looping = createLooping(lastSink);
-
- if (!currentPattern.getQuantifier().hasProperty(QuantifierProperty.OPTIONAL)) {
- lastSink = createFirstMandatoryStateOfLoop(looping);
- } else if (currentPattern instanceof FollowedByPattern &&
- currentPattern.getQuantifier().hasProperty(QuantifierProperty.CONSECUTIVE)) {
- lastSink = createWaitingStateForZeroOrMore(looping, lastSink);
- } else {
- lastSink = looping;
- }
- } else if (currentPattern.getQuantifier().hasProperty(QuantifierProperty.TIMES)) {
- lastSink = createTimesState(lastSink, currentPattern.getTimes());
- } else {
- lastSink = createSingletonState(lastSink);
- }
+ lastSink = convertPattern(lastSink);
currentPattern = currentPattern.getPrevious();
final Time currentWindowTime = currentPattern.getWindowTime();
@@ -190,6 +170,29 @@ public class NFACompiler {
return lastSink;
}
+ private State<T> convertPattern(final State<T> sinkState) {
+ final State<T> lastSink;
+ checkPatternNameUniqueness();
+ usedNames.add(currentPattern.getName());
+
+ final Quantifier quantifier = currentPattern.getQuantifier();
+ if (quantifier.hasProperty(Quantifier.QuantifierProperty.LOOPING)) {
+ final State<T> looping = createLooping(sinkState);
+
+ if (!quantifier.hasProperty(Quantifier.QuantifierProperty.OPTIONAL)) {
+ lastSink = createFirstMandatoryStateOfLoop(looping);
+ } else {
+ lastSink = createWaitingStateForZeroOrMore(looping, sinkState);
+ }
+ } else if (quantifier.hasProperty(Quantifier.QuantifierProperty.TIMES)) {
+ lastSink = createTimesState(sinkState, currentPattern.getTimes());
+ } else {
+ lastSink = createSingletonState(sinkState);
+ }
+
+ return lastSink;
+ }
+
/**
* Creates a pair of states that enables relaxed strictness before a zeroOrMore looping state.
*
@@ -197,19 +200,21 @@ public class NFACompiler {
* @param lastSink the state that the looping one points to
* @return the newly created state
*/
+ @SuppressWarnings("unchecked")
private State<T> createWaitingStateForZeroOrMore(final State<T> loopingState, final State<T> lastSink) {
- final State<T> followByState = createNormalState();
- final State<T> followByStateWithoutProceed = createNormalState();
-
final IterativeCondition<T> currentFunction = (IterativeCondition<T>)currentPattern.getCondition();
- final IterativeCondition<T> ignoreFunction = getIgnoreCondition(currentPattern);
+ final State<T> followByState = createNormalState();
followByState.addProceed(lastSink, BooleanConditions.<T>trueFunction());
- followByState.addIgnore(followByStateWithoutProceed, ignoreFunction);
followByState.addTake(loopingState, currentFunction);
- followByStateWithoutProceed.addIgnore(ignoreFunction);
- followByStateWithoutProceed.addTake(loopingState, currentFunction);
+ final IterativeCondition<T> ignoreFunction = getIgnoreCondition(currentPattern);
+ if (ignoreFunction != null) {
+ final State<T> followByStateWithoutProceed = createNormalState();
+ followByState.addIgnore(followByStateWithoutProceed, ignoreFunction);
+ followByStateWithoutProceed.addIgnore(ignoreFunction);
+ followByStateWithoutProceed.addTake(loopingState, currentFunction);
+ }
return followByState;
}
@@ -230,23 +235,7 @@ public class NFACompiler {
*/
@SuppressWarnings("unchecked")
private State<T> createStartState(State<T> sinkState) {
- checkPatternNameUniqueness();
- usedNames.add(currentPattern.getName());
-
- final State<T> beginningState;
- if (currentPattern.getQuantifier().hasProperty(QuantifierProperty.LOOPING)) {
- final State<T> loopingState = createLooping(sinkState);
- if (!currentPattern.getQuantifier().hasProperty(QuantifierProperty.OPTIONAL)) {
- beginningState = createFirstMandatoryStateOfLoop(loopingState);
- } else {
- beginningState = loopingState;
- }
- } else if (currentPattern.getQuantifier().hasProperty(QuantifierProperty.TIMES)) {
- beginningState = createTimesState(sinkState, currentPattern.getTimes());
- } else {
- beginningState = createSingletonState(sinkState);
- }
-
+ final State<T> beginningState = convertPattern(sinkState);
beginningState.makeStart();
return beginningState;
@@ -263,29 +252,26 @@ public class NFACompiler {
private State<T> createTimesState(final State<T> sinkState, int times) {
State<T> lastSink = sinkState;
for (int i = 0; i < times - 1; i++) {
- lastSink = createSingletonState(
- lastSink,
- !currentPattern.getQuantifier().hasProperty(QuantifierProperty.CONSECUTIVE),
- false);
+ lastSink = createSingletonState(lastSink, getInnerIgnoreCondition(currentPattern), false);
}
+ final IterativeCondition<T> currentFilterFunction = (IterativeCondition<T>) currentPattern.getCondition();
+ final IterativeCondition<T> ignoreCondition = getIgnoreCondition(currentPattern);
+
// we created the intermediate states in the loop, now we create the start of the loop.
- if (!currentPattern.getQuantifier().hasProperty(QuantifierProperty.OPTIONAL)) {
- return createSingletonState(lastSink, currentPattern instanceof FollowedByPattern, false);
+ if (!currentPattern.getQuantifier().hasProperty(Quantifier.QuantifierProperty.OPTIONAL)) {
+ return createSingletonState(lastSink, ignoreCondition, false);
}
- final IterativeCondition<T> currentFilterFunction = (IterativeCondition<T>) currentPattern.getCondition();
- final IterativeCondition<T> trueFunction = BooleanConditions.trueFunction();
-
final State<T> singletonState = createNormalState();
singletonState.addTake(lastSink, currentFilterFunction);
- singletonState.addProceed(sinkState, trueFunction);
+ singletonState.addProceed(sinkState, BooleanConditions.<T>trueFunction());
- if (currentPattern instanceof FollowedByPattern) {
+ if (ignoreCondition != null) {
State<T> ignoreState = createNormalState();
ignoreState.addTake(lastSink, currentFilterFunction);
- ignoreState.addIgnore(trueFunction);
- singletonState.addIgnore(ignoreState, trueFunction);
+ ignoreState.addIgnore(ignoreCondition);
+ singletonState.addIgnore(ignoreState, ignoreCondition);
}
return singletonState;
}
@@ -300,8 +286,10 @@ public class NFACompiler {
*/
@SuppressWarnings("unchecked")
private State<T> createSingletonState(final State<T> sinkState) {
- return createSingletonState(sinkState, currentPattern instanceof FollowedByPattern,
- currentPattern.getQuantifier().hasProperty(QuantifierProperty.OPTIONAL));
+ return createSingletonState(
+ sinkState,
+ getIgnoreCondition(currentPattern),
+ currentPattern.getQuantifier().hasProperty(Quantifier.QuantifierProperty.OPTIONAL));
}
/**
@@ -309,12 +297,12 @@ public class NFACompiler {
* of a similar state without the PROCEED edge, so that for each PROCEED transition branches
* in computation state graph can be created only once.
*
- * @param addIgnore if any IGNORE should be added
+ * @param ignoreCondition condition that should be applied to IGNORE transition
* @param sinkState state that the state being converted should point to
* @return the created state
*/
@SuppressWarnings("unchecked")
- private State<T> createSingletonState(final State<T> sinkState, boolean addIgnore, boolean isOptional) {
+ private State<T> createSingletonState(final State<T> sinkState, final IterativeCondition<T> ignoreCondition, final boolean isOptional) {
final IterativeCondition<T> currentFilterFunction = (IterativeCondition<T>) currentPattern.getCondition();
final IterativeCondition<T> trueFunction = BooleanConditions.trueFunction();
@@ -325,7 +313,7 @@ public class NFACompiler {
singletonState.addProceed(sinkState, trueFunction);
}
- if (addIgnore) {
+ if (ignoreCondition != null) {
final State<T> ignoreState;
if (isOptional) {
ignoreState = createNormalState();
@@ -333,7 +321,7 @@ public class NFACompiler {
} else {
ignoreState = singletonState;
}
- singletonState.addIgnore(ignoreState, getIgnoreCondition(currentPattern));
+ singletonState.addIgnore(ignoreState, ignoreCondition);
}
return singletonState;
}
@@ -352,8 +340,8 @@ public class NFACompiler {
final State<T> firstState = createNormalState();
firstState.addTake(sinkState, currentFilterFunction);
- if (currentPattern instanceof FollowedByPattern) {
- final IterativeCondition<T> ignoreCondition = getIgnoreCondition(currentPattern);
+ final IterativeCondition<T> ignoreCondition = getIgnoreCondition(currentPattern);
+ if (ignoreCondition != null) {
firstState.addIgnore(ignoreCondition);
}
return firstState;
@@ -369,18 +357,16 @@ public class NFACompiler {
*/
@SuppressWarnings("unchecked")
private State<T> createLooping(final State<T> sinkState) {
-
- final State<T> loopingState = createNormalState();
final IterativeCondition<T> filterFunction = (IterativeCondition<T>) currentPattern.getCondition();
+ final IterativeCondition<T> ignoreCondition = getInnerIgnoreCondition(currentPattern);
final IterativeCondition<T> trueFunction = BooleanConditions.trueFunction();
+ final State<T> loopingState = createNormalState();
loopingState.addProceed(sinkState, trueFunction);
loopingState.addTake(filterFunction);
- if (!currentPattern.getQuantifier().hasProperty(QuantifierProperty.CONSECUTIVE)) {
- final State<T> ignoreState = createNormalState();
-
- final IterativeCondition<T> ignoreCondition = getIgnoreCondition(currentPattern);
+ if (ignoreCondition != null) {
+ final State<T> ignoreState = createNormalState();
ignoreState.addTake(loopingState, filterFunction);
ignoreState.addIgnore(ignoreCondition);
loopingState.addIgnore(ignoreState, ignoreCondition);
@@ -403,15 +389,37 @@ public class NFACompiler {
/**
* @return The {@link IterativeCondition condition} for the {@code IGNORE} edge
- * that corresponds to the specified {@link Pattern}. If the pattern is
- * {@link QuantifierProperty#EAGER}, the negated user-specified condition is
- * returned. In other case, a condition that always evaluated to {@code true} is
- * returned.
+ * that corresponds to the specified {@link Pattern}. It is applicable only for inner states of a complex
+ * state like looping or times.
*/
+ @SuppressWarnings("unchecked")
+ private IterativeCondition<T> getInnerIgnoreCondition(Pattern<T, ?> pattern) {
+ switch (pattern.getQuantifier().getInnerConsumingStrategy()) {
+ case STRICT:
+ return null;
+ case SKIP_TILL_NEXT:
+ return new NotCondition<>((IterativeCondition<T>) pattern.getCondition());
+ case SKIP_TILL_ANY:
+ return BooleanConditions.trueFunction();
+ }
+ return null;
+ }
+
+ /**
+ * @return The {@link IterativeCondition condition} for the {@code IGNORE} edge
+ * that corresponds to the specified {@link Pattern}. For more on strategy see {@link Quantifier}
+ */
+ @SuppressWarnings("unchecked")
private IterativeCondition<T> getIgnoreCondition(Pattern<T, ?> pattern) {
- return pattern.getQuantifier().hasProperty(QuantifierProperty.EAGER)
- ? new NotCondition<>((IterativeCondition<T>) pattern.getCondition())
- : BooleanConditions.<T>trueFunction();
+ switch (pattern.getQuantifier().getConsumingStrategy()) {
+ case STRICT:
+ return null;
+ case SKIP_TILL_NEXT:
+ return new NotCondition<>((IterativeCondition<T>) pattern.getCondition());
+ case SKIP_TILL_ANY:
+ return BooleanConditions.trueFunction();
+ }
+ return null;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/FollowedByPattern.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/FollowedByPattern.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/FollowedByPattern.java
deleted file mode 100644
index 266451c..0000000
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/FollowedByPattern.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.cep.pattern;
-
-/**
- * Pattern operator which signifies that the there is a non-strict temporal contiguity between
- * itself and its preceding pattern operator. This means that there might be events in between
- * two matching events. These events are then simply ignored.
- *
- * @param <T> Base type of the events
- * @param <F> Subtype of T to which the operator is currently constrained
- */
-public class FollowedByPattern<T, F extends T> extends Pattern<T, F> {
- FollowedByPattern(final String name, Pattern<T, ?> previous) {
- super(name, previous);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java
index 3ce0c73..cef0f85 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java
@@ -20,6 +20,7 @@ package org.apache.flink.cep.pattern;
import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.cep.nfa.NFA;
+import org.apache.flink.cep.pattern.Quantifier.ConsumingStrategy;
import org.apache.flink.cep.pattern.conditions.AndCondition;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.OrCondition;
@@ -56,8 +57,8 @@ public class Pattern<T, F extends T> {
/** Window length in which the pattern match has to occur. */
private Time windowTime;
- /** A quantifier for the pattern. By default set to {@link Quantifier#ONE()}. */
- private Quantifier quantifier = Quantifier.ONE();
+ /** A quantifier for the pattern. By default set to {@link Quantifier#ONE(ConsumingStrategy)}. */
+ private Quantifier quantifier = Quantifier.ONE(ConsumingStrategy.STRICT);
/**
* Applicable to a {@code times} pattern, and holds
@@ -70,6 +71,15 @@ public class Pattern<T, F extends T> {
this.previous = previous;
}
+ protected Pattern(
+ final String name,
+ final Pattern<T, ? extends T> previous,
+ final ConsumingStrategy consumingStrategy) {
+ this.name = name;
+ this.previous = previous;
+ this.quantifier = Quantifier.ONE(consumingStrategy);
+ }
+
public Pattern<T, ? extends T> getPrevious() {
return previous;
}
@@ -195,7 +205,19 @@ public class Pattern<T, F extends T> {
* @return A new pattern which is appended to this one
*/
public Pattern<T, T> next(final String name) {
- return new Pattern<T, T>(name, this);
+ return new Pattern<T, T>(name, this, ConsumingStrategy.STRICT);
+ }
+
+ /**
+ * Appends a new pattern to the existing one. The new pattern enforces non-strict
+ * temporal contiguity. This means that a matching event of this pattern and the
+ * preceding matching event might be interleaved with other events which are ignored.
+ *
+ * @param name Name of the new pattern
+ * @return A new pattern which is appended to this one
+ */
+ public Pattern<T, T> followedBy(final String name) {
+ return new Pattern<>(name, this, ConsumingStrategy.SKIP_TILL_NEXT);
}
/**
@@ -206,8 +228,8 @@ public class Pattern<T, F extends T> {
* @param name Name of the new pattern
* @return A new pattern which is appended to this one
*/
- public FollowedByPattern<T, T> followedBy(final String name) {
- return new FollowedByPattern<T, T>(name, this);
+ public Pattern<T, T> followedByAny(final String name) {
+ return new Pattern<>(name, this, ConsumingStrategy.SKIP_TILL_ANY);
}
/**
@@ -232,12 +254,12 @@ public class Pattern<T, F extends T> {
* {@code A1 A2 B} appears, this will generate patterns:
* {@code A1 B} and {@code A1 A2 B}. See also {@link #allowCombinations()}.
*
- * @return The same pattern with a {@link Quantifier#ONE_OR_MORE()} quantifier applied.
+ * @return The same pattern with a {@link Quantifier#ONE_OR_MORE(ConsumingStrategy)} quantifier applied.
* @throws MalformedPatternException if the quantifier is not applicable to this pattern.
*/
public Pattern<T, F> oneOrMore() {
checkIfQuantifierApplied();
- this.quantifier = Quantifier.ONE_OR_MORE();
+ this.quantifier = Quantifier.ONE_OR_MORE(quantifier.getConsumingStrategy());
return this;
}
@@ -252,14 +274,14 @@ public class Pattern<T, F extends T> {
public Pattern<T, F> times(int times) {
checkIfQuantifierApplied();
Preconditions.checkArgument(times > 0, "You should give a positive number greater than 0.");
- this.quantifier = Quantifier.TIMES();
+ this.quantifier = Quantifier.TIMES(quantifier.getConsumingStrategy());
this.times = times;
return this;
}
/**
- * Applicable only to {@link Quantifier#ONE_OR_MORE()} and {@link Quantifier#TIMES()} patterns,
- * this option allows more flexibility to the matching events.
+ * Applicable only to {@link Quantifier#ONE_OR_MORE(ConsumingStrategy)} and
+ * {@link Quantifier#TIMES(ConsumingStrategy)} patterns, this option allows more flexibility to the matching events.
*
* <p>If {@code allowCombinations()} is not applied for a
* pattern {@code A.oneOrMore().followedBy(B)} and a sequence of events
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java
index 0332ed4..b0f882c 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java
@@ -24,48 +24,62 @@ public class Quantifier {
private final EnumSet<QuantifierProperty> properties;
- private Quantifier(final QuantifierProperty first, final QuantifierProperty... rest) {
+ private final ConsumingStrategy consumingStrategy;
+
+ private ConsumingStrategy innerConsumingStrategy = ConsumingStrategy.SKIP_TILL_NEXT;
+
+ private Quantifier(
+ final ConsumingStrategy consumingStrategy,
+ final QuantifierProperty first,
+ final QuantifierProperty... rest) {
this.properties = EnumSet.of(first, rest);
+ this.consumingStrategy = consumingStrategy;
}
- public static Quantifier ONE() {
- return new Quantifier(QuantifierProperty.SINGLE);
+ public static Quantifier ONE(final ConsumingStrategy consumingStrategy) {
+ return new Quantifier(consumingStrategy, QuantifierProperty.SINGLE);
}
- public static Quantifier ONE_OR_MORE() {
- return new Quantifier(QuantifierProperty.LOOPING, QuantifierProperty.EAGER);
+ public static Quantifier ONE_OR_MORE(final ConsumingStrategy consumingStrategy) {
+ return new Quantifier(consumingStrategy, QuantifierProperty.LOOPING);
}
- public static Quantifier TIMES() {
- return new Quantifier(QuantifierProperty.TIMES, QuantifierProperty.EAGER);
+ public static Quantifier TIMES(final ConsumingStrategy consumingStrategy) {
+ return new Quantifier(consumingStrategy, QuantifierProperty.TIMES);
}
public boolean hasProperty(QuantifierProperty property) {
return properties.contains(property);
}
- public void combinations() {
- if (!hasProperty(QuantifierProperty.SINGLE) && !hasProperty(Quantifier.QuantifierProperty.EAGER)) {
- throw new MalformedPatternException("Combinations already allowed!");
- }
+ public ConsumingStrategy getConsumingStrategy() {
+ return consumingStrategy;
+ }
- if (hasProperty(Quantifier.QuantifierProperty.LOOPING) || hasProperty(Quantifier.QuantifierProperty.TIMES)) {
- properties.remove(Quantifier.QuantifierProperty.EAGER);
- } else {
- throw new MalformedPatternException("Combinations not applicable to " + this + "!");
+ public ConsumingStrategy getInnerConsumingStrategy() {
+ return innerConsumingStrategy;
+ }
+
+ private static void checkPattern(boolean condition, Object errorMessage) {
+ if (!condition) {
+ throw new MalformedPatternException(String.valueOf(errorMessage));
}
}
+ public void combinations() {
+ checkPattern(!hasProperty(QuantifierProperty.SINGLE), "Combinations not applicable to " + this + "!");
+ checkPattern(innerConsumingStrategy != ConsumingStrategy.STRICT, "You can apply apply either combinations or consecutive, not both!");
+ checkPattern(innerConsumingStrategy != ConsumingStrategy.SKIP_TILL_ANY, "Combinations already applied!");
+
+ innerConsumingStrategy = ConsumingStrategy.SKIP_TILL_ANY;
+ }
+
public void consecutive() {
- if (!hasProperty(QuantifierProperty.SINGLE) && hasProperty(Quantifier.QuantifierProperty.CONSECUTIVE)) {
- throw new MalformedPatternException("Strict continuity already applied!");
- }
+ checkPattern(hasProperty(QuantifierProperty.LOOPING) || hasProperty(QuantifierProperty.TIMES), "Combinations not applicable to " + this + "!");
+ checkPattern(innerConsumingStrategy != ConsumingStrategy.SKIP_TILL_ANY, "You can apply apply either combinations or consecutive, not both!");
+ checkPattern(innerConsumingStrategy != ConsumingStrategy.STRICT, "Combinations already applied!");
- if (hasProperty(Quantifier.QuantifierProperty.LOOPING) || hasProperty(Quantifier.QuantifierProperty.TIMES)) {
- properties.add(Quantifier.QuantifierProperty.CONSECUTIVE);
- } else {
- throw new MalformedPatternException("Strict continuity not applicable to " + this + "!");
- }
+ innerConsumingStrategy = ConsumingStrategy.STRICT;
}
public void optional() {
@@ -76,13 +90,21 @@ public class Quantifier {
}
@Override
- public boolean equals(Object obj) {
- return obj instanceof Quantifier && properties.equals(((Quantifier)obj).properties);
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Quantifier that = (Quantifier) o;
+ return Objects.equals(properties, that.properties) &&
+ consumingStrategy == that.consumingStrategy;
}
@Override
public int hashCode() {
- return Objects.hashCode(properties);
+ return Objects.hash(properties, consumingStrategy);
}
/**
@@ -92,9 +114,13 @@ public class Quantifier {
SINGLE,
LOOPING,
TIMES,
- EAGER,
- CONSECUTIVE,
OPTIONAL
}
+ public enum ConsumingStrategy {
+ STRICT,
+ SKIP_TILL_NEXT,
+ SKIP_TILL_ANY
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7c35dc0e/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
index 3a32175..f62c686 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CEPITCase.java
@@ -96,7 +96,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
return value.getName().equals("start");
}
})
- .followedBy("middle").subtype(SubEvent.class).where(
+ .followedByAny("middle").subtype(SubEvent.class).where(
new SimpleCondition<SubEvent>() {
@Override
@@ -105,7 +105,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
}
}
)
- .followedBy("end").where(new SimpleCondition<Event>() {
+ .followedByAny("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
@@ -171,7 +171,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
return value.getName().equals("start");
}
})
- .followedBy("middle").subtype(SubEvent.class).where(
+ .followedByAny("middle").subtype(SubEvent.class).where(
new SimpleCondition<SubEvent>() {
@Override
@@ -180,7 +180,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
}
}
)
- .followedBy("end").where(new SimpleCondition<Event>() {
+ .followedByAny("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
@@ -250,13 +250,13 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("middle");
}
- }).followedBy("end").where(new SimpleCondition<Event>() {
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
@@ -339,13 +339,13 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("middle");
}
- }).followedBy("end").where(new SimpleCondition<Event>() {
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
@@ -416,7 +416,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
DataStream<Integer> input = env.fromElements(1, 2);
- Pattern<Integer, ?> pattern = Pattern.<Integer>begin("start").followedBy("end").within(Time.days(1));
+ Pattern<Integer, ?> pattern = Pattern.<Integer>begin("start").followedByAny("end").within(Time.days(1));
DataStream<Integer> result = CEP.pattern(input, pattern).select(new PatternSelectFunction<Integer, Integer>() {
@Override
@@ -470,13 +470,13 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("middle");
}
- }).followedBy("end").where(new SimpleCondition<Event>() {
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
@@ -538,7 +538,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
return value.getName().equals("start");
}
})
- .followedBy("middle")
+ .followedByAny("middle")
.where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
@@ -551,7 +551,7 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
return value.getPrice() == 5.0;
}
})
- .followedBy("end").where(new SimpleCondition<Event>() {
+ .followedByAny("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
@@ -623,13 +623,13 @@ public class CEPITCase extends StreamingMultipleProgramsTestBase {
public boolean filter(Event value) throws Exception {
return value.getName().equals("start");
}
- }).followedBy("middle").where(new SimpleCondition<Event>() {
+ }).followedByAny("middle").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("middle");
}
- }).followedBy("end").where(new SimpleCondition<Event>() {
+ }).followedByAny("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {