@Test public void testCepOperatorWindowTime() { final CepEventPattern<CepExampleClass> event1 = new CepEventPattern.Builder<CepExampleClass>() .setName("first") .setCondition(conditionA) .setClass(exampleClassType) .setContiguity(strictContiguity) .build(); final CepEventPattern<CepExampleClass> event2 = new CepEventPattern.Builder<CepExampleClass>() .setName("second") .setCondition(conditionA) .setClass(exampleClassType) .setContiguity(strictContiguity) .build(); final long exampleWindowTime = 10L; final List<CepEventPattern<CepExampleClass>> exampleEventSequence = new ArrayList<>(); exampleEventSequence.add(event1); exampleEventSequence.add(event2); final CepExampleClass value1 = new CepExampleClass("A", 1); final CepExampleClass value2 = new CepExampleClass("A", 2); final CepExampleClass value3 = new CepExampleClass("A", 3); final CepExampleClass value4 = new CepExampleClass("A", 4); final MistDataEvent data1 = new MistDataEvent(value1, 1L); final MistDataEvent data2 = new MistDataEvent(value2, 9L); final MistDataEvent data3 = new MistDataEvent(value3, 20L); final MistDataEvent data4 = new MistDataEvent(value4, 23L); final CepOperator cepOperator = new CepOperator(exampleEventSequence, exampleWindowTime); final List<MistEvent> result = new LinkedList<>(); cepOperator.setOutputEmitter(new OutputBufferEmitter(result)); cepOperator.processLeftData(data1); cepOperator.processLeftData(data2); cepOperator.processLeftData(data3); cepOperator.processLeftData(data4); Assert.assertEquals(2, result.size()); final CepExampleClass a01 = getCepExampleClass(result, 0, "first", 0); final CepExampleClass a02 = getCepExampleClass(result, 0, "second", 0); Assert.assertEquals("A", a01.getName()); Assert.assertEquals(1, a01.getAge()); Assert.assertEquals("A", a02.getName()); Assert.assertEquals(2, a02.getAge()); final CepExampleClass a13 = getCepExampleClass(result, 1, "first", 0); final CepExampleClass a14 = getCepExampleClass(result, 1, "second", 0); Assert.assertEquals("A", a13.getName()); Assert.assertEquals(3, a13.getAge()); Assert.assertEquals("A", a14.getName()); Assert.assertEquals(4, a14.getAge()); }
@Override public void processLeftData(final MistDataEvent data) { final T input = (T) data.getValue(); final long timeStamp = data.getTimestamp(); final List<EventStack<T>> newMatchedEventStackList = new ArrayList<>(); for (int iterStackIndex = 0; iterStackIndex < matchedEventStackList.size(); iterStackIndex++) { final EventStack<T> stack = matchedEventStackList.get(iterStackIndex); boolean isDiscard = true; if (stack.getFirstEventTime() + windowTime >= timeStamp) { final int stateIndex = stack.getStack().peek().getIndex(); final int minProceedIndex = (int) proceedIndexList.get(stateIndex).get(0); final int maxProceedIndex = (int) proceedIndexList.get(stateIndex).get(1); if (minProceedIndex == -1 && maxProceedIndex == -1) { continue; } final CepEventPattern<T> currEventPattern = eventPatternList.get(stateIndex); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { if (proceedIndex == stateIndex && proceedIndex != 0) { if (currEventPattern.isRepeated() && !stack.getStack().peek().isStopped()) { final int times = stack.getStack().peek().getlist().size(); if (currEventPattern.getStopCondition().test(input)) { stack.getStack().peek().setStopped(); if (times < currEventPattern.getMinRepetition()) { continue; } } else if (currEventPattern.getCondition().test(input)) { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } if (currEventPattern.getMaxRepetition() == -1 || times < currEventPattern.getMaxRepetition()) { final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); newStack.getStack().peek().addEvent(input); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex && newStack.isEmitted()) { emit(data, newStack); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } else { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT) { stack.getStack().peek().setStopped(); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.RELAXED || currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } else { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); final int times = stack.getStack().peek().getlist().size(); if (cepEventPattern.getCondition().test(input)) { if (cepEventPattern.getContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } else { if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } } if (!isDiscard) { final EventStack<T> newStack = stack.deepCopy(); newStack.setIncludeLast(false); if (!stack.isEmitted()) { newStack.setAlreadyEmitted(); } newMatchedEventStackList.add(newStack); } } final int minProceedIndex = (int) proceedIndexList.get(0).get(0); final int maxProceedIndex = (int) proceedIndexList.get(0).get(1); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); if (cepEventPattern.getCondition().test(input)) { final EventStack<T> newStack = new EventStack<>(timeStamp); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } } } matchedEventStackList.clear(); matchedEventStackList.addAll(newMatchedEventStackList); }
CepOperator extends OneStreamOperator { @Override public void processLeftData(final MistDataEvent data) { final T input = (T) data.getValue(); final long timeStamp = data.getTimestamp(); final List<EventStack<T>> newMatchedEventStackList = new ArrayList<>(); for (int iterStackIndex = 0; iterStackIndex < matchedEventStackList.size(); iterStackIndex++) { final EventStack<T> stack = matchedEventStackList.get(iterStackIndex); boolean isDiscard = true; if (stack.getFirstEventTime() + windowTime >= timeStamp) { final int stateIndex = stack.getStack().peek().getIndex(); final int minProceedIndex = (int) proceedIndexList.get(stateIndex).get(0); final int maxProceedIndex = (int) proceedIndexList.get(stateIndex).get(1); if (minProceedIndex == -1 && maxProceedIndex == -1) { continue; } final CepEventPattern<T> currEventPattern = eventPatternList.get(stateIndex); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { if (proceedIndex == stateIndex && proceedIndex != 0) { if (currEventPattern.isRepeated() && !stack.getStack().peek().isStopped()) { final int times = stack.getStack().peek().getlist().size(); if (currEventPattern.getStopCondition().test(input)) { stack.getStack().peek().setStopped(); if (times < currEventPattern.getMinRepetition()) { continue; } } else if (currEventPattern.getCondition().test(input)) { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } if (currEventPattern.getMaxRepetition() == -1 || times < currEventPattern.getMaxRepetition()) { final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); newStack.getStack().peek().addEvent(input); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex && newStack.isEmitted()) { emit(data, newStack); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } else { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT) { stack.getStack().peek().setStopped(); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.RELAXED || currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } else { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); final int times = stack.getStack().peek().getlist().size(); if (cepEventPattern.getCondition().test(input)) { if (cepEventPattern.getContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } else { if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } } if (!isDiscard) { final EventStack<T> newStack = stack.deepCopy(); newStack.setIncludeLast(false); if (!stack.isEmitted()) { newStack.setAlreadyEmitted(); } newMatchedEventStackList.add(newStack); } } final int minProceedIndex = (int) proceedIndexList.get(0).get(0); final int maxProceedIndex = (int) proceedIndexList.get(0).get(1); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); if (cepEventPattern.getCondition().test(input)) { final EventStack<T> newStack = new EventStack<>(timeStamp); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } } } matchedEventStackList.clear(); matchedEventStackList.addAll(newMatchedEventStackList); } }
CepOperator extends OneStreamOperator { @Override public void processLeftData(final MistDataEvent data) { final T input = (T) data.getValue(); final long timeStamp = data.getTimestamp(); final List<EventStack<T>> newMatchedEventStackList = new ArrayList<>(); for (int iterStackIndex = 0; iterStackIndex < matchedEventStackList.size(); iterStackIndex++) { final EventStack<T> stack = matchedEventStackList.get(iterStackIndex); boolean isDiscard = true; if (stack.getFirstEventTime() + windowTime >= timeStamp) { final int stateIndex = stack.getStack().peek().getIndex(); final int minProceedIndex = (int) proceedIndexList.get(stateIndex).get(0); final int maxProceedIndex = (int) proceedIndexList.get(stateIndex).get(1); if (minProceedIndex == -1 && maxProceedIndex == -1) { continue; } final CepEventPattern<T> currEventPattern = eventPatternList.get(stateIndex); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { if (proceedIndex == stateIndex && proceedIndex != 0) { if (currEventPattern.isRepeated() && !stack.getStack().peek().isStopped()) { final int times = stack.getStack().peek().getlist().size(); if (currEventPattern.getStopCondition().test(input)) { stack.getStack().peek().setStopped(); if (times < currEventPattern.getMinRepetition()) { continue; } } else if (currEventPattern.getCondition().test(input)) { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } if (currEventPattern.getMaxRepetition() == -1 || times < currEventPattern.getMaxRepetition()) { final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); newStack.getStack().peek().addEvent(input); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex && newStack.isEmitted()) { emit(data, newStack); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } else { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT) { stack.getStack().peek().setStopped(); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.RELAXED || currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } else { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); final int times = stack.getStack().peek().getlist().size(); if (cepEventPattern.getCondition().test(input)) { if (cepEventPattern.getContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } else { if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } } if (!isDiscard) { final EventStack<T> newStack = stack.deepCopy(); newStack.setIncludeLast(false); if (!stack.isEmitted()) { newStack.setAlreadyEmitted(); } newMatchedEventStackList.add(newStack); } } final int minProceedIndex = (int) proceedIndexList.get(0).get(0); final int maxProceedIndex = (int) proceedIndexList.get(0).get(1); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); if (cepEventPattern.getCondition().test(input)) { final EventStack<T> newStack = new EventStack<>(timeStamp); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } } } matchedEventStackList.clear(); matchedEventStackList.addAll(newMatchedEventStackList); } @Inject private  CepOperator(
      @Parameter(CepEventPatterns.class) final String serializedEvents,
      @Parameter(WindowTime.class) final long windowTime,
      final ClassLoader classLoader);  CepOperator(
      final List<CepEventPattern<T>> cepEventPatterns,
      final long windowTime); }
CepOperator extends OneStreamOperator { @Override public void processLeftData(final MistDataEvent data) { final T input = (T) data.getValue(); final long timeStamp = data.getTimestamp(); final List<EventStack<T>> newMatchedEventStackList = new ArrayList<>(); for (int iterStackIndex = 0; iterStackIndex < matchedEventStackList.size(); iterStackIndex++) { final EventStack<T> stack = matchedEventStackList.get(iterStackIndex); boolean isDiscard = true; if (stack.getFirstEventTime() + windowTime >= timeStamp) { final int stateIndex = stack.getStack().peek().getIndex(); final int minProceedIndex = (int) proceedIndexList.get(stateIndex).get(0); final int maxProceedIndex = (int) proceedIndexList.get(stateIndex).get(1); if (minProceedIndex == -1 && maxProceedIndex == -1) { continue; } final CepEventPattern<T> currEventPattern = eventPatternList.get(stateIndex); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { if (proceedIndex == stateIndex && proceedIndex != 0) { if (currEventPattern.isRepeated() && !stack.getStack().peek().isStopped()) { final int times = stack.getStack().peek().getlist().size(); if (currEventPattern.getStopCondition().test(input)) { stack.getStack().peek().setStopped(); if (times < currEventPattern.getMinRepetition()) { continue; } } else if (currEventPattern.getCondition().test(input)) { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } if (currEventPattern.getMaxRepetition() == -1 || times < currEventPattern.getMaxRepetition()) { final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); newStack.getStack().peek().addEvent(input); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex && newStack.isEmitted()) { emit(data, newStack); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } else { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT) { stack.getStack().peek().setStopped(); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.RELAXED || currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } else { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); final int times = stack.getStack().peek().getlist().size(); if (cepEventPattern.getCondition().test(input)) { if (cepEventPattern.getContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } else { if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } } if (!isDiscard) { final EventStack<T> newStack = stack.deepCopy(); newStack.setIncludeLast(false); if (!stack.isEmitted()) { newStack.setAlreadyEmitted(); } newMatchedEventStackList.add(newStack); } } final int minProceedIndex = (int) proceedIndexList.get(0).get(0); final int maxProceedIndex = (int) proceedIndexList.get(0).get(1); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); if (cepEventPattern.getCondition().test(input)) { final EventStack<T> newStack = new EventStack<>(timeStamp); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } } } matchedEventStackList.clear(); matchedEventStackList.addAll(newMatchedEventStackList); } @Inject private  CepOperator(
      @Parameter(CepEventPatterns.class) final String serializedEvents,
      @Parameter(WindowTime.class) final long windowTime,
      final ClassLoader classLoader);  CepOperator(
      final List<CepEventPattern<T>> cepEventPatterns,
      final long windowTime); @Override void processLeftData(final MistDataEvent data); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override void processLeftCheckpoint(final MistCheckpointEvent input); }
CepOperator extends OneStreamOperator { @Override public void processLeftData(final MistDataEvent data) { final T input = (T) data.getValue(); final long timeStamp = data.getTimestamp(); final List<EventStack<T>> newMatchedEventStackList = new ArrayList<>(); for (int iterStackIndex = 0; iterStackIndex < matchedEventStackList.size(); iterStackIndex++) { final EventStack<T> stack = matchedEventStackList.get(iterStackIndex); boolean isDiscard = true; if (stack.getFirstEventTime() + windowTime >= timeStamp) { final int stateIndex = stack.getStack().peek().getIndex(); final int minProceedIndex = (int) proceedIndexList.get(stateIndex).get(0); final int maxProceedIndex = (int) proceedIndexList.get(stateIndex).get(1); if (minProceedIndex == -1 && maxProceedIndex == -1) { continue; } final CepEventPattern<T> currEventPattern = eventPatternList.get(stateIndex); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { if (proceedIndex == stateIndex && proceedIndex != 0) { if (currEventPattern.isRepeated() && !stack.getStack().peek().isStopped()) { final int times = stack.getStack().peek().getlist().size(); if (currEventPattern.getStopCondition().test(input)) { stack.getStack().peek().setStopped(); if (times < currEventPattern.getMinRepetition()) { continue; } } else if (currEventPattern.getCondition().test(input)) { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } if (currEventPattern.getMaxRepetition() == -1 || times < currEventPattern.getMaxRepetition()) { final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); newStack.getStack().peek().addEvent(input); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex && newStack.isEmitted()) { emit(data, newStack); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } else { if (currEventPattern.getInnerContiguity() == CepEventContiguity.STRICT) { stack.getStack().peek().setStopped(); } if (currEventPattern.getInnerContiguity() == CepEventContiguity.RELAXED || currEventPattern.getInnerContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } else { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); final int times = stack.getStack().peek().getlist().size(); if (cepEventPattern.getCondition().test(input)) { if (cepEventPattern.getContiguity() == CepEventContiguity.STRICT && !stack.isIncludingLast()) { continue; } final EventStack<T> newStack = new EventStack<>(stack.getFirstEventTime()); newStack.setStack(stack.deepCopy().getStack()); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } else { if (cepEventPattern.getContiguity() == CepEventContiguity.NON_DETERMINISTIC_RELAXED) { isDiscard = false; } } } } } if (!isDiscard) { final EventStack<T> newStack = stack.deepCopy(); newStack.setIncludeLast(false); if (!stack.isEmitted()) { newStack.setAlreadyEmitted(); } newMatchedEventStackList.add(newStack); } } final int minProceedIndex = (int) proceedIndexList.get(0).get(0); final int maxProceedIndex = (int) proceedIndexList.get(0).get(1); for (int proceedIndex = minProceedIndex; proceedIndex <= maxProceedIndex; proceedIndex++) { final CepEventPattern<T> cepEventPattern = eventPatternList.get(proceedIndex); if (cepEventPattern.getCondition().test(input)) { final EventStack<T> newStack = new EventStack<>(timeStamp); final EventStackEntry<T> newEntry = new EventStackEntry<>(proceedIndex); newEntry.addEvent(input); newStack.getStack().push(newEntry); newMatchedEventStackList.add(newStack); if (proceedIndex >= minFinalStateIndex) { emit(data, newStack); } } } matchedEventStackList.clear(); matchedEventStackList.addAll(newMatchedEventStackList); } @Inject private  CepOperator(
      @Parameter(CepEventPatterns.class) final String serializedEvents,
      @Parameter(WindowTime.class) final long windowTime,
      final ClassLoader classLoader);  CepOperator(
      final List<CepEventPattern<T>> cepEventPatterns,
      final long windowTime); @Override void processLeftData(final MistDataEvent data); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override void processLeftCheckpoint(final MistCheckpointEvent input);  }