code: @PublicEvolving
  	public <ACC, V, R> SingleOutputStreamOperator<R> aggregate(
  			AggregateFunction<T, ACC, V> aggFunction,
  			ProcessAllWindowFunction<V, R, W> windowFunction) {
  
  		checkNotNull(aggFunction, "aggFunction");
  		checkNotNull(windowFunction, "windowFunction");
  
  		TypeInformation<ACC> accumulatorType = TypeExtractor.getAggregateFunctionAccumulatorType(
  				aggFunction, input.getType(), null, false);
  
  		TypeInformation<V> aggResultType = TypeExtractor.getAggregateFunctionReturnType(
  				aggFunction, input.getType(), null, false);
  
  		TypeInformation<R> resultType = getProcessAllWindowFunctionReturnType(windowFunction, aggResultType);
  
  		return aggregate(aggFunction, windowFunction, accumulatorType, aggResultType, resultType);
  	}
docs: Applies the given window function to each window. The window function is called for each
  evaluation of the window for each key individually. The output of the window function is
  interpreted as a regular non-windowed stream.
  
  <p>Arriving data is incrementally aggregated using the given aggregate function. This means
  that the window function typically has only a single value to process when called.
  
  @param aggFunction The aggregate function that is used for incremental aggregation.
  @param windowFunction The process window function.
  
  @return The data stream that is the result of applying the window function to the window.
  
  @param <ACC> The type of the AggregateFunction's accumulator
  @param <V> The type of AggregateFunction's result, and the WindowFunction's input
  @param <R> The type of the elements in the resulting stream, equal to the
  WindowFunction's result type