@Test public void testConditionalBranchOperator() throws InjectionException { final MistDataEvent d1 = new MistDataEvent("1", 1L); final MistDataEvent d2 = new MistDataEvent("22", 2L); final MistDataEvent d3 = new MistDataEvent("333", 3L); final MistDataEvent d4 = new MistDataEvent("4444", 4L); final MistDataEvent d5 = new MistDataEvent("55555", 5L); final MistWatermarkEvent w1 = new MistWatermarkEvent(6L); final List<MISTPredicate<String>> predicates = new ArrayList<>(); predicates.add((input) -> input.length() < 2); predicates.add((input) -> input.length() < 3); predicates.add((input) -> input.length() < 5); final List<Tuple<MistEvent, Integer>> result = new LinkedList<>(); final ConditionalBranchOperator<String> conditionalBranchOperator = new ConditionalBranchOperator<>(predicates); conditionalBranchOperator.setOutputEmitter(new IndexOutputEmitter(result)); conditionalBranchOperator.processLeftData(d1); conditionalBranchOperator.processLeftData(d2); conditionalBranchOperator.processLeftData(d3); conditionalBranchOperator.processLeftData(d4); conditionalBranchOperator.processLeftData(d5); conditionalBranchOperator.processLeftWatermark(w1); Assert.assertEquals(5, result.size()); Assert.assertEquals(new Tuple<>(d1, 1), result.get(0)); Assert.assertEquals(new Tuple<>(d2, 2), result.get(1)); Assert.assertEquals(new Tuple<>(d3, 3), result.get(2)); Assert.assertEquals(new Tuple<>(d4, 3), result.get(3)); Assert.assertEquals(new Tuple<>(w1, 0), result.get(4)); }
@Inject private ConditionalBranchOperator( @Parameter(SerializedUdfList.class) final List<String> serializedUdfList, final ClassLoader classLoader) throws IOException, ClassNotFoundException { predicates = new ArrayList<>(serializedUdfList.size()); for (final String serializedUdf : serializedUdfList) { predicates.add(SerializeUtils.deserializeFromString(serializedUdf, classLoader)); } }
ConditionalBranchOperator extends OneStreamOperator { @Inject private ConditionalBranchOperator( @Parameter(SerializedUdfList.class) final List<String> serializedUdfList, final ClassLoader classLoader) throws IOException, ClassNotFoundException { predicates = new ArrayList<>(serializedUdfList.size()); for (final String serializedUdf : serializedUdfList) { predicates.add(SerializeUtils.deserializeFromString(serializedUdf, classLoader)); } } }
ConditionalBranchOperator extends OneStreamOperator { @Inject private ConditionalBranchOperator( @Parameter(SerializedUdfList.class) final List<String> serializedUdfList, final ClassLoader classLoader) throws IOException, ClassNotFoundException { predicates = new ArrayList<>(serializedUdfList.size()); for (final String serializedUdf : serializedUdfList) { predicates.add(SerializeUtils.deserializeFromString(serializedUdf, classLoader)); } } @Inject private  ConditionalBranchOperator(
      @Parameter(SerializedUdfList.class) final List<String> serializedUdfList,
      final ClassLoader classLoader);  ConditionalBranchOperator(final List<MISTPredicate<I>> predicates); }
ConditionalBranchOperator extends OneStreamOperator { @Inject private ConditionalBranchOperator( @Parameter(SerializedUdfList.class) final List<String> serializedUdfList, final ClassLoader classLoader) throws IOException, ClassNotFoundException { predicates = new ArrayList<>(serializedUdfList.size()); for (final String serializedUdf : serializedUdfList) { predicates.add(SerializeUtils.deserializeFromString(serializedUdf, classLoader)); } } @Inject private  ConditionalBranchOperator(
      @Parameter(SerializedUdfList.class) final List<String> serializedUdfList,
      final ClassLoader classLoader);  ConditionalBranchOperator(final List<MISTPredicate<I>> predicates); @Override void processLeftData(final MistDataEvent input); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override void processLeftCheckpoint(final MistCheckpointEvent input); }
ConditionalBranchOperator extends OneStreamOperator { @Inject private ConditionalBranchOperator( @Parameter(SerializedUdfList.class) final List<String> serializedUdfList, final ClassLoader classLoader) throws IOException, ClassNotFoundException { predicates = new ArrayList<>(serializedUdfList.size()); for (final String serializedUdf : serializedUdfList) { predicates.add(SerializeUtils.deserializeFromString(serializedUdf, classLoader)); } } @Inject private  ConditionalBranchOperator(
      @Parameter(SerializedUdfList.class) final List<String> serializedUdfList,
      final ClassLoader classLoader);  ConditionalBranchOperator(final List<MISTPredicate<I>> predicates); @Override void processLeftData(final MistDataEvent input); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override void processLeftCheckpoint(final MistCheckpointEvent input);  }