ReduceWindow extends ShuffleOperator<InputT, Byte, OutputT> implements TypeAware.Value<ValueT>, CompositeOperator<InputT, OutputT> { public static <InputT> ValueByReduceByBuilder<InputT, InputT> of(PCollection<InputT> input) { return named(null).of(input); } private  ReduceWindow(
      @Nullable String name,
      UnaryFunction<InputT, ValueT> valueExtractor,
      @Nullable TypeDescriptor<ValueT> valueType,
      ReduceFunctor<ValueT, OutputT> reducer,
      @Nullable BinaryFunction<ValueT, ValueT, Integer> valueComparator,
      @Nullable Window<InputT> window,
      TypeDescriptor<OutputT> outputType); private  ReduceWindow(
      @Nullable String name,
      UnaryFunction<InputT, ValueT> valueExtractor,
      @Nullable TypeDescriptor<ValueT> valueType,
      VoidFunction<AccT> accumulatorFactory,
      BinaryFunction<AccT, ValueT, AccT> accumulate,
      CombinableBinaryFunction<AccT> mergeAccumulators,
      UnaryFunction<AccT, OutputT> outputFn,
      @Nullable TypeDescriptor<AccT> accumulatorType,
      @Nullable BinaryFunction<ValueT, ValueT, Integer> valueComparator,
      @Nullable Window<InputT> window,
      TypeDescriptor<OutputT> outputType); static ValueByReduceByBuilder<InputT, InputT> of(PCollection<InputT> input); static OfBuilder named(@Nullable String name); boolean isCombineFnStyle(); ReduceFunctor<ValueT, OutputT> getReducer(); boolean isCombinable(); UnaryFunction<InputT, ValueT> getValueExtractor(); Optional<BinaryFunction<ValueT, ValueT, Integer>> getValueComparator(); @Override Optional<TypeDescriptor<ValueT>> getValueType(); @Override @SuppressWarnings("unchecked") PCollection<OutputT> expand(PCollectionList<InputT> inputs);  }
@Test public void testWindow_applyIf() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<Long> output = ReduceWindow.of(dataset) .reduceBy(e -> 1L) .withSortedValues(String::compareTo) .applyIf( true, b -> b.windowBy(FixedWindows.of(org.joda.time.Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .discardingFiredPanes()) .output(); final ReduceWindow rw = (ReduceWindow) TestUtils.getProducer(output); assertTrue(rw.getWindow().isPresent()); @SuppressWarnings("unchecked") final WindowDesc<?> windowDesc = WindowDesc.of((Window) rw.getWindow().get()); assertEquals( FixedWindows.of(org.joda.time.Duration.standardHours(1)), windowDesc.getWindowFn()); assertEquals(DefaultTrigger.of(), windowDesc.getTrigger()); assertEquals(AccumulationMode.DISCARDING_FIRED_PANES, windowDesc.getAccumulationMode()); }