NFA implements Serializable { public static <K, V> NFA<K, V> build(final Stages<K, V> stages, final AggregatesStore<K> aggregatesStore, final SharedVersionedBufferStore<K, V> sharedVersionedBuffer) { LinkedBlockingQueue<ComputationStage<K, V>> queue = new LinkedBlockingQueue<>(); queue.add(stages.initialComputationStage()); return new NFA<>(aggregatesStore, sharedVersionedBuffer, stages.getDefinedStates(), queue); } NFA(final AggregatesStore<K> aggregatesStore,
                final SharedVersionedBufferStore<K, V> sharedVersionedBuffer,
                final Set<String> aggregatesName,
                final Queue<ComputationStage<K, V>> computationStages);  NFA(final AggregatesStore<K> aggregatesStore,
               final SharedVersionedBufferStore<K, V> sharedVersionedBuffer,
               final Set<String> aggregatesName,
               final Queue<ComputationStage<K, V>> computationStages, final long runs); static NFA<K, V> build(final Stages<K, V> stages,
                                         final AggregatesStore<K> aggregatesStore,
                                         final SharedVersionedBufferStore<K, V> sharedVersionedBuffer); long getRuns(); Queue<ComputationStage<K, V>> getComputationStages(); List<Sequence<K, V>> matchPattern(final Event<K, V> event);  }
@Test public void testNFAGivenStatefulCondition() { Pattern<String, Integer> pattern = new QueryBuilder<String, Integer>() .select("first") .where(TestMatcher.isGreaterThan(0)) .fold("sum", (key, value, state) -> value) .fold("count", (key, value, state) -> 1) .then() .select("second") .oneOrMore() .where((event, states) -> { double average =(int)states.get("sum") / (int)states.get("count"); return average >= event.value(); }) .<Integer>fold("sum", (key, value, state) -> state + value) .<Integer>fold("count", (key, value, state) -> state + 1) .then() .select("latest") .where((event, states) -> { double average = (int)states.get("sum") / (int)states.get("count"); return average < event.value(); }) .build(); final NFA<String, Integer> nfa = newNFA(pattern); Event<String, Integer> e1 = nextEvent("t1", "key", 5); Event<String, Integer> e2 = nextEvent("t1", "key", 3); Event<String, Integer> e3 = nextEvent("t1", "key", 4); Event<String, Integer> e4 = nextEvent("t1", "key", 10); List<Sequence<String, Integer>> s = simulate(nfa, e1, e2, e3, e4); assertEquals(1, s.size()); assertNFA(nfa, 5, 2); Sequence<String, Integer> expected = Sequence.<String, Integer>newBuilder() .add("latest", e4) .add("second", e3) .add("second", e2) .add("first", e1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenSequenceCondition() { Pattern<String, Integer> pattern = new QueryBuilder<String, Integer>() .select("first") .where(TestMatcher.isGreaterThan(0)) .then() .select("second") .oneOrMore() .where((event, sequence, states) -> { double average = StreamSupport.stream(sequence.spliterator(), false) .mapToInt(Event::value) .summaryStatistics() .getAverage(); return average >= event.value(); }) .then() .select("latest") .where((event, sequence, states) -> { double average = StreamSupport.stream(sequence.spliterator(), false) .mapToInt(Event::value) .summaryStatistics() .getAverage(); return average < event.value(); }) .build(); final NFA<String, Integer> nfa = newNFA(pattern); Event<String, Integer> e1 = nextEvent("t1", "key", 5); Event<String, Integer> e2 = nextEvent("t1", "key", 3); Event<String, Integer> e3 = nextEvent("t1", "key", 4); Event<String, Integer> e4 = nextEvent("t1", "key", 10); List<Sequence<String, Integer>> s = simulate(nfa, e1, e2, e3, e4); assertEquals(1, s.size()); assertNFA(nfa, 5, 2); Sequence<String, Integer> expected = Sequence.<String, Integer>newBuilder() .add("latest", e4) .add("second", e3) .add("second", e2) .add("first", e1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenExpectingOccurrencesStage() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .times(3) .where(TestMatcher.isEqualTo("C")) .then() .select("latest") .where(TestMatcher.isEqualTo("E")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev3, ev4, ev6, ev8); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev8) .add("second", ev6) .add("second", ev4) .add("second", ev3) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenZeroOrMoreStageWhenNoMatchingInputs() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .zeroOrMore() .where(TestMatcher.isEqualTo("C")) .then() .select("latest") .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev5); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev5) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenZeroOrMoreStageWhenMatchingInputs() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .zeroOrMore() .where(TestMatcher.isEqualTo("C")) .then() .select("latest") .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev3, ev4, ev5); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev5) .add("second", ev4) .add("second", ev3) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenOptionalExpectingOccurrenceStageWhenNoMatchingInputs() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .times(2) .optional() .where(TestMatcher.isEqualTo("C")) .then() .select("latest") .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev5); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev5) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenOptionalExpectingOccurrenceStageWhenMatchingInputs() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .times(2) .optional() .where(TestMatcher.isEqualTo("C")) .then() .select("latest") .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev3, ev4, ev5); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev5) .add("second", ev4) .add("second", ev3) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenExpectingOccurrencesStageWhenSkipTilNextMatchContiguityIsUsed() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second", Selected.withSkipTilNextMatch()) .times(3) .where(TestMatcher.isEqualTo("C")) .then() .select("latest") .where(TestMatcher.isEqualTo("E")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev3, ev4, ev5, ev6, ev8); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev8) .add("second", ev6) .add("second", ev4) .add("second", ev3) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenOptionalStageAndStrictContiguity() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .optional() .where(TestMatcher.isEqualTo("B")) .then() .select("latest") .where(TestMatcher.isEqualTo("C")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev3); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev3) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenOneRunAndStrictContiguity() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .where(TestMatcher.isEqualTo("B")) .then() .select("latest") .where(TestMatcher.isEqualTo("C")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("latest", ev3) .add("second", ev2) .add("first", ev1) .build(true); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenOneRunAndMultipleMatch() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("firstStage") .where(TestMatcher.isEqualTo("A")) .then() .select("secondStage") .where(TestMatcher.isEqualTo("B")) .then() .select("thirdStage") .oneOrMore() .where(TestMatcher.isEqualTo("C")) .then() .select("latestState") .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3, ev4, ev5); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("firstStage", ev1) .add("secondStage", ev2) .add("thirdStage", ev3) .add("thirdStage", ev4) .add("latestState", ev5) .build(false); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenTwoConsecutiveSkipTillNextMatch() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second", Selected.withSkipTilNextMatch()) .where(TestMatcher.isEqualTo("C")) .then() .select("latest", Selected.withSkipTilNextMatch()) .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3, ev4, ev5); assertEquals(1, s.size()); assertNFA(nfa, 2, 1); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev3) .add("latest", ev5) .build(false); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenTwoConsecutiveSkipTillNextMatchAndMultipleMatch() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second", Selected.withSkipTilNextMatch()) .oneOrMore() .where(TestMatcher.isEqualTo("C")) .then() .select("latest", Selected.withSkipTilNextMatch()) .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3, ev4, ev5); assertEquals(1, s.size()); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev3) .add("second", ev4) .add("latest", ev5) .build(false); assertEquals(expected, s.get(0)); }
@Test public void testNFAGivenTwoConsecutiveSkipTillAnyMatch() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second", Selected.withSkipTilAnyMatch()) .where(TestMatcher.isEqualTo("C")) .then() .select("latest", Selected.withSkipTilAnyMatch()) .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3, ev4, ev5); assertNFA(nfa, 6, 4); assertEquals(2, s.size()); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev3) .add("latest", ev5) .build(false); assertEquals(expected, s.get(0)); Sequence<String, String> expected2 = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev4) .add("latest", ev5) .build(false); assertEquals(expected2, s.get(1)); }
@Test public void testNFAGivenMultipleMatchAndSkipTillAnyMatch() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second", Selected.withSkipTilAnyMatch()) .oneOrMore() .where(TestMatcher.isEqualTo("C")) .then() .select("latest") .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3, ev4, ev5); assertNFA(nfa, 5, 2); assertEquals(3, s.size()); Sequence<String, String> expected = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev3) .add("second", ev4) .add("latest", ev5) .build(false); assertEquals(expected, s.get(0)); Sequence<String, String> expected2 = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev3) .add("latest", ev5) .build(false); assertEquals(expected2, s.get(1)); Sequence<String, String> expected3 = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev4) .add("latest", ev5) .build(false); assertEquals(expected3, s.get(2)); }
@Test public void testNFAGivenTwoConsecutiveSkipTilAnyMatch() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .where(TestMatcher.isEqualTo("B")) .then() .select("three", Selected.withSkipTilAnyMatch()) .where(TestMatcher.isEqualTo("C")) .then() .select("latest", Selected.withSkipTilAnyMatch()) .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3, ev4, ev5); assertNFA(nfa, 6, 4); assertEquals(2, s.size()); Sequence<String, String> expected1 = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev2) .add("three", ev3) .add("latest", ev5) .build(false); assertEquals(expected1, s.get(0)); Sequence<String, String> expected2 = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev2) .add("three", ev4) .add("latest", ev5) .build(false); assertEquals(expected2, s.get(1)); }
@Test public void testNFAGivenMultipleStrategies() { Pattern<String, String> pattern = new QueryBuilder<String, String>() .select("first") .where(TestMatcher.isEqualTo("A")) .then() .select("second") .where(TestMatcher.isEqualTo("B")) .then() .select("three", Selected.withSkipTilAnyMatch()) .where(TestMatcher.isEqualTo("C")) .then() .select("latest", Selected.withSkipTilNextMatch()) .where(TestMatcher.isEqualTo("D")) .build(); final NFA<String, String> nfa = newNFA(pattern); List<Sequence<String, String>> s = simulate(nfa, ev1, ev2, ev3, ev4, ev5); assertNFA(nfa, 4, 2); assertEquals(2, s.size()); Sequence<String, String> expected1 = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev2) .add("three", ev3) .add("latest", ev5) .build(false); assertEquals(expected1, s.get(0)); Sequence<String, String> expected2 = Sequence.<String, String>newBuilder() .add("first", ev1) .add("second", ev2) .add("three", ev4) .add("latest", ev5) .build(false); assertEquals(expected2, s.get(1)); }