TriggerStateMachines { public static TriggerStateMachine stateMachineForTrigger(RunnerApi.Trigger trigger) { switch (trigger.getTriggerCase()) { case AFTER_ALL: return AfterAllStateMachine.of( stateMachinesForTriggers(trigger.getAfterAll().getSubtriggersList())); case AFTER_ANY: return AfterFirstStateMachine.of( stateMachinesForTriggers(trigger.getAfterAny().getSubtriggersList())); case AFTER_END_OF_WINDOW: return stateMachineForAfterEndOfWindow(trigger.getAfterEndOfWindow()); case ELEMENT_COUNT: return AfterPaneStateMachine.elementCountAtLeast( trigger.getElementCount().getElementCount()); case AFTER_SYNCHRONIZED_PROCESSING_TIME: return AfterSynchronizedProcessingTimeStateMachine.ofFirstElement(); case DEFAULT: return DefaultTriggerStateMachine.of(); case NEVER: return NeverStateMachine.ever(); case ALWAYS: return ReshuffleTriggerStateMachine.create(); case OR_FINALLY: return stateMachineForTrigger(trigger.getOrFinally().getMain()) .orFinally(stateMachineForTrigger(trigger.getOrFinally().getFinally())); case REPEAT: return RepeatedlyStateMachine.forever( stateMachineForTrigger(trigger.getRepeat().getSubtrigger())); case AFTER_EACH: return AfterEachStateMachine.inOrder( stateMachinesForTriggers(trigger.getAfterEach().getSubtriggersList())); case AFTER_PROCESSING_TIME: return stateMachineForAfterProcessingTime(trigger.getAfterProcessingTime()); case TRIGGER_NOT_SET: throw new IllegalArgumentException( String.format("Required field 'trigger' not set on %s", trigger)); default: throw new IllegalArgumentException(String.format("Unknown trigger type %s", trigger)); } } private  TriggerStateMachines(); static TriggerStateMachine stateMachineForTrigger(RunnerApi.Trigger trigger);  }
@Test public void testStateMachineForAfterPane() { int count = 37; RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setElementCount(RunnerApi.Trigger.ElementCount.newBuilder().setElementCount(count)) .build(); AfterPaneStateMachine machine = (AfterPaneStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat(machine.getElementCount(), equalTo(trigger.getElementCount().getElementCount())); }
@Test public void testStateMachineForAfterWatermark() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setAfterEndOfWindow(RunnerApi.Trigger.AfterEndOfWindow.getDefaultInstance()) .build(); AfterWatermarkStateMachine.FromEndOfWindow machine = (AfterWatermarkStateMachine.FromEndOfWindow) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat( TriggerStateMachines.stateMachineForTrigger(trigger), instanceOf(AfterWatermarkStateMachine.FromEndOfWindow.class)); }
@Test public void testDefaultTriggerTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setDefault(RunnerApi.Trigger.Default.getDefaultInstance()) .build(); assertThat( TriggerStateMachines.stateMachineForTrigger(trigger), instanceOf(DefaultTriggerStateMachine.class)); }
@Test public void testNeverTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setNever(RunnerApi.Trigger.Never.getDefaultInstance()) .build(); NeverStateMachine machine = (NeverStateMachine) checkNotNull(TriggerStateMachines.stateMachineForTrigger(trigger)); }
@Test public void testAfterEachTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setAfterEach( RunnerApi.Trigger.AfterEach.newBuilder() .addSubtriggers(subtrigger1) .addSubtriggers(subtrigger2)) .build(); AfterEachStateMachine machine = (AfterEachStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat(machine, equalTo(AfterEachStateMachine.inOrder(submachine1, submachine2))); }
@Test public void testAfterFirstTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setAfterAny( RunnerApi.Trigger.AfterAny.newBuilder() .addSubtriggers(subtrigger1) .addSubtriggers(subtrigger2)) .build(); AfterFirstStateMachine machine = (AfterFirstStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat(machine, equalTo(AfterFirstStateMachine.of(submachine1, submachine2))); }
@Test public void testAfterAllTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setAfterAll( RunnerApi.Trigger.AfterAll.newBuilder() .addSubtriggers(subtrigger1) .addSubtriggers(subtrigger2)) .build(); AfterAllStateMachine machine = (AfterAllStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat(machine, equalTo(AfterAllStateMachine.of(submachine1, submachine2))); }
@Test public void testAfterWatermarkEarlyTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setAfterEndOfWindow( RunnerApi.Trigger.AfterEndOfWindow.newBuilder().setEarlyFirings(subtrigger1)) .build(); AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate machine = (AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat( machine, equalTo(AfterWatermarkStateMachine.pastEndOfWindow().withEarlyFirings(submachine1))); }
@Test public void testAfterWatermarkEarlyLateTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setAfterEndOfWindow( RunnerApi.Trigger.AfterEndOfWindow.newBuilder() .setEarlyFirings(subtrigger1) .setLateFirings(subtrigger2)) .build(); AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate machine = (AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat( machine, equalTo( AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(submachine1) .withLateFirings(submachine2))); }
@Test public void testOrFinallyTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setOrFinally( RunnerApi.Trigger.OrFinally.newBuilder() .setMain(subtrigger1) .setFinally(subtrigger2)) .build(); OrFinallyStateMachine machine = (OrFinallyStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat(machine, equalTo(submachine1.orFinally(submachine2))); }
@Test public void testRepeatedlyTranslation() { RunnerApi.Trigger trigger = RunnerApi.Trigger.newBuilder() .setRepeat(RunnerApi.Trigger.Repeat.newBuilder().setSubtrigger(subtrigger1)) .build(); RepeatedlyStateMachine machine = (RepeatedlyStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger); assertThat(machine, equalTo(RepeatedlyStateMachine.forever(submachine1))); }