CombineFnUtil { public static <InputT, AccumT, OutputT> CombineFnWithContext<InputT, AccumT, OutputT> toFnWithContext( GlobalCombineFn<InputT, AccumT, OutputT> globalCombineFn) { if (globalCombineFn instanceof CombineFnWithContext) { @SuppressWarnings("unchecked") CombineFnWithContext<InputT, AccumT, OutputT> combineFnWithContext = (CombineFnWithContext<InputT, AccumT, OutputT>) globalCombineFn; return combineFnWithContext; } else { @SuppressWarnings("unchecked") final CombineFn<InputT, AccumT, OutputT> combineFn = (CombineFn<InputT, AccumT, OutputT>) globalCombineFn; return new CombineFnWithContext<InputT, AccumT, OutputT>() { @Override public AccumT createAccumulator(Context c) { return combineFn.createAccumulator(); } @Override public AccumT addInput(AccumT accumulator, InputT input, Context c) { return combineFn.addInput(accumulator, input); } @Override public AccumT mergeAccumulators(Iterable<AccumT> accumulators, Context c) { return combineFn.mergeAccumulators(accumulators); } @Override public OutputT extractOutput(AccumT accumulator, Context c) { return combineFn.extractOutput(accumulator); } @Override public AccumT compact(AccumT accumulator, Context c) { return combineFn.compact(accumulator); } @Override public OutputT defaultValue() { return combineFn.defaultValue(); } @Override public Coder<AccumT> getAccumulatorCoder(CoderRegistry registry, Coder<InputT> inputCoder) throws CannotProvideCoderException { return combineFn.getAccumulatorCoder(registry, inputCoder); } @Override public Coder<OutputT> getDefaultOutputCoder( CoderRegistry registry, Coder<InputT> inputCoder) throws CannotProvideCoderException { return combineFn.getDefaultOutputCoder(registry, inputCoder); } @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); combineFn.populateDisplayData(builder); } }; } }  static CombineFn<InputT, AccumT, OutputT> bindContext(
      CombineFnWithContext<InputT, AccumT, OutputT> combineFn, StateContext<?> stateContext); static CombineFnWithContext<InputT, AccumT, OutputT> toFnWithContext(
          GlobalCombineFn<InputT, AccumT, OutputT> globalCombineFn);  }
@Test public void testToFnWithContextIdempotent() throws Exception { CombineFnWithContext<Integer, int[], Integer> fnWithContext = CombineFnUtil.toFnWithContext(Sum.ofIntegers()); assertTrue(fnWithContext == CombineFnUtil.toFnWithContext(fnWithContext)); }
@Test public void testToFnWithContext() throws Exception { CombineFnWithContext<Integer, int[], Integer> fnWithContext = CombineFnUtil.toFnWithContext(Sum.ofIntegers()); List<Integer> inputs = ImmutableList.of(1, 2, 3, 4); Context nullContext = CombineContextFactory.nullContext(); int[] accum = fnWithContext.createAccumulator(nullContext); for (Integer i : inputs) { accum = fnWithContext.addInput(accum, i, nullContext); } assertEquals(10, fnWithContext.extractOutput(accum, nullContext).intValue()); }