code: @PublicEvolving
  	public <ACC, R> SingleOutputStreamOperator<R> aggregate(AggregateFunction<T, ACC, R> function) {
  		checkNotNull(function, "function");
  
  		if (function instanceof RichFunction) {
  			throw new UnsupportedOperationException("This aggregation function cannot be a RichFunction.");
  		}
  
  		TypeInformation<ACC> accumulatorType = TypeExtractor.getAggregateFunctionAccumulatorType(
  				function, input.getType(), null, false);
  
  		TypeInformation<R> resultType = TypeExtractor.getAggregateFunctionReturnType(
  				function, input.getType(), null, false);
  
  		return aggregate(function, accumulatorType, resultType);
  	}
docs: Applies the given {@code AggregateFunction} to each window. The AggregateFunction
  aggregates all elements of a window into a single result element. The stream of these
  result elements (one per window) is interpreted as a regular non-windowed stream.
  
  @param function The aggregation function.
  @return The data stream that is the result of applying the fold function to the window.
  
  @param <ACC> The type of the AggregateFunction's accumulator
  @param <R> The type of the elements in the resulting stream, equal to the
  AggregateFunction's result type