StateTransitionOperator extends OneStreamStateHandlerOperator { @Inject private StateTransitionOperator( @Parameter(InitialState.class) final String initialState, @Parameter(FinalState.class) final Set<String> finalState, @Parameter(StateTable.class) final String stateTable, final ClassLoader classLoader) throws IOException, ClassNotFoundException { this( initialState, finalState, SerializeUtils.deserializeFromString(stateTable, classLoader)); } @Inject private  StateTransitionOperator(
      @Parameter(InitialState.class) final String initialState,
      @Parameter(FinalState.class) final Set<String> finalState,
      @Parameter(StateTable.class) final String stateTable,
      final ClassLoader classLoader);  StateTransitionOperator(
      final String initialState,
      final Set<String> finalState,
      final Map<String, Collection<Tuple2<MISTPredicate, String>>> stateTable); @Override void processLeftData(final MistDataEvent input); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override Map<String, Object> getStateSnapshot(); @Override void setState(final Map<String, Object> loadedState); @Override void processLeftCheckpoint(final MistCheckpointEvent input);  }
@Test public void testStateTransitionOperator() throws InterruptedException { final Map<String, Integer> value1 = new HashMap<>(); value1.put("number", 1); final MistDataEvent data1 = new MistDataEvent(value1, 0L); final Map<String, Integer> value2 = new HashMap<>(); value2.put("number", 2); final MistDataEvent data2 = new MistDataEvent(value2, 1L); final MistWatermarkEvent watermarkEvent = new MistWatermarkEvent(7L); final Map<String, Integer> value3 = new HashMap<>(); value3.put("number", 3); final MistDataEvent data3 = new MistDataEvent(value3, 10L); final Map<String, Integer> value4 = new HashMap<>(); value4.put("number", 4); final MistDataEvent data4 = new MistDataEvent(value4, 11L); final Set<String> finalSet = new HashSet<>(); finalSet.add("1"); finalSet.add("4"); final Map<String, Collection<Tuple2<MISTPredicate, String>>> stateTable = new HashMap<>(); final Collection<Tuple2<MISTPredicate, String>> list0 = new ArrayList<>(); list0.add(new Tuple2<>(new RuleBasedEQPredicate("number", 1), "1")); list0.add(new Tuple2<>(new RuleBasedEQPredicate("number", 3), "3")); final Collection<Tuple2<MISTPredicate, String>> list1 = new ArrayList<>(); list1.add(new Tuple2<>(new RuleBasedEQPredicate("number", 2), "0")); final Collection<Tuple2<MISTPredicate, String>> list3 = new ArrayList<>(); list3.add(new Tuple2<>(new RuleBasedEQPredicate("number", 4), "4")); stateTable.put("0", list0); stateTable.put("1", list1); stateTable.put("3", list3); final StateTransitionOperator stateTransitionOperator = new StateTransitionOperator("0", finalSet, stateTable); final List<MistEvent> result = new LinkedList<>(); stateTransitionOperator.setOutputEmitter(new OutputBufferEmitter(result)); stateTransitionOperator.processLeftData(data1); Assert.assertEquals(1, result.size()); Assert.assertEquals(data1, result.get(0)); Assert.assertEquals("1", stateTransitionOperator.getStateSnapshot().get("stateTransitionOperatorState")); stateTransitionOperator.processLeftData(data2); Assert.assertEquals(1, result.size()); Assert.assertEquals("0", stateTransitionOperator.getStateSnapshot().get("stateTransitionOperatorState")); stateTransitionOperator.processLeftWatermark(watermarkEvent); Assert.assertEquals(2, result.size()); Assert.assertEquals(watermarkEvent, result.get(1)); Assert.assertEquals("0", stateTransitionOperator.getStateSnapshot().get("stateTransitionOperatorState")); stateTransitionOperator.processLeftData(data3); Assert.assertEquals(2, result.size()); Assert.assertEquals("3", stateTransitionOperator.getStateSnapshot().get("stateTransitionOperatorState")); stateTransitionOperator.processLeftData(data4); Assert.assertEquals(3, result.size()); Assert.assertEquals(data4, result.get(2)); Assert.assertEquals("4", stateTransitionOperator.getStateSnapshot().get("stateTransitionOperatorState")); }