OrFinallyStateMachine extends TriggerStateMachine { @Override public boolean shouldFire(TriggerStateMachine.TriggerContext context) throws Exception { return context.trigger().subTrigger(ACTUAL).invokeShouldFire(context) || context.trigger().subTrigger(UNTIL).invokeShouldFire(context); } @VisibleForTesting  OrFinallyStateMachine(TriggerStateMachine actual, TriggerStateMachine until); @Override void onElement(OnElementContext c); @Override void onMerge(OnMergeContext c); @Override boolean shouldFire(TriggerStateMachine.TriggerContext context); @Override void onFire(TriggerStateMachine.TriggerContext context); @Override String toString();  }
@Test public void testActualFiresAndFinishes() throws Exception { tester = TriggerStateMachineTester.forTrigger( new OrFinallyStateMachine( AfterPaneStateMachine.elementCountAtLeast(2), AfterPaneStateMachine.elementCountAtLeast(100)), FixedWindows.of(Duration.millis(100))); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100)); tester.injectElements(1); assertFalse(tester.shouldFire(window)); assertFalse(tester.isMarkedFinished(window)); tester.injectElements(2); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertTrue(tester.isMarkedFinished(window)); }
@Test public void testActualFiresOnly() throws Exception { tester = TriggerStateMachineTester.forTrigger( new OrFinallyStateMachine( RepeatedlyStateMachine.forever(AfterPaneStateMachine.elementCountAtLeast(2)), AfterPaneStateMachine.elementCountAtLeast(100)), FixedWindows.of(Duration.millis(100))); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100)); tester.injectElements(1); assertFalse(tester.shouldFire(window)); assertFalse(tester.isMarkedFinished(window)); tester.injectElements(2); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertFalse(tester.isMarkedFinished(window)); tester.injectElements(3, 4); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertFalse(tester.isMarkedFinished(window)); }
@Test public void testShouldFireAfterMerge() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterEachStateMachine.inOrder( AfterPaneStateMachine.elementCountAtLeast(5) .orFinally(AfterWatermarkStateMachine.pastEndOfWindow()), RepeatedlyStateMachine.forever(AfterPaneStateMachine.elementCountAtLeast(1))), Sessions.withGapDuration(Duration.millis(10))); tester.injectElements(1); IntervalWindow firstWindow = new IntervalWindow(new Instant(1), new Instant(11)); assertFalse(tester.shouldFire(firstWindow)); tester.advanceInputWatermark(new Instant(11)); assertTrue(tester.shouldFire(firstWindow)); tester.fireIfShouldFire(firstWindow); tester.injectElements(5); IntervalWindow secondWindow = new IntervalWindow(new Instant(5), new Instant(15)); assertFalse(tester.shouldFire(secondWindow)); tester.mergeWindows(); IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(15)); assertFalse(tester.shouldFire(mergedWindow)); tester.injectElements(1, 2, 3, 4, 5); tester.mergeWindows(); assertTrue(tester.shouldFire(mergedWindow)); }
@Test public void testActualFiresButUntilFinishes() throws Exception { tester = TriggerStateMachineTester.forTrigger( new OrFinallyStateMachine( RepeatedlyStateMachine.forever(AfterPaneStateMachine.elementCountAtLeast(2)), AfterPaneStateMachine.elementCountAtLeast(3)), FixedWindows.of(Duration.millis(10))); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(10)); tester.injectElements(1); assertFalse(tester.shouldFire(window)); assertFalse(tester.isMarkedFinished(window)); tester.injectElements(2); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertFalse(tester.isMarkedFinished(window)); tester.injectElements(3); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertTrue(tester.isMarkedFinished(window)); }