/*
 * Copyright 2015 Data Artisans GmbH
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.dataartisans.flink.dataflow;

import com.google.cloud.dataflow.sdk.PipelineResult;
import java.util.Collections;
import java.util.Map;

/**
 * Result of executing a {@link com.google.cloud.dataflow.sdk.Pipeline} with Flink. This
 * has methods to query to job runtime and the final values of
 * {@link com.google.cloud.dataflow.sdk.transforms.Aggregator}s.
 */
public class FlinkRunnerResult implements PipelineResult {
	
	private final Map<String, Object> aggregators;
	
	private final long runtime;
	
	public FlinkRunnerResult(Map<String, Object> aggregators, long runtime) {
		this.aggregators = (aggregators == null || aggregators.isEmpty()) ?
				Collections.<String, Object>emptyMap() :
				Collections.unmodifiableMap(aggregators);
		
		this.runtime = runtime;
	}

	/**
	 * Return the final values of all {@link com.google.cloud.dataflow.sdk.transforms.Aggregator}s
	 * used in the {@link com.google.cloud.dataflow.sdk.Pipeline}.
	 */
	public Map<String, Object> getAggregators() {
		return aggregators;
	}

	public long getRuntime() {
		return runtime;
	}
}

@@ -690,16 +690,297 @@
 Result;%0A
+import com.google.cloud.dataflow.sdk.runners.AggregatorRetrievalException;%0Aimport com.google.cloud.dataflow.sdk.runners.AggregatorValues;%0Aimport com.google.cloud.dataflow.sdk.transforms.Aggregator;%0Aimport org.apache.flink.runtime.messages.accumulators.AccumulatorResultsNotFound;%0A%0A
 import j
@@ -1665,96 +1665,290 @@
 %7D%0A%0A%09
-/**%0A%09 * Return the final values of all %7B@link com.google.cloud.dataflow.sdk.transf
+@Override%0A%09public State getState() %7B%0A%09%09return null;%0A%09%7D%0A%0A%09@Override%0A%09public %3CT%3E AggregatorValues%3CT%3E getAggregatorValues(final Aggregator%3C?, T%3E aggregator) throws AggregatorRetrievalException %7B%0A%09%09// TODO provide a list of all accumulator step values%0A%09%09Object value = aggregat
 or
-m
 s.
-A
+get(a
 ggre
@@ -1956,127 +1956,146 @@
 ator
-%7Ds%0A%09 * used in the %7B@link com.google.cloud.dataflow.sdk.Pipeline%7D.%0A%09 */%0A%09public Map%3CString, Object%3E getAggregator
+.getName());%0A%09%09if (value != null) %7B%0A%09%09%09return new AggregatorValues%3CT%3E() %7B%0A%09%09%09%09@Override%0A%09%09%09%09public Map%3CString, T%3E getValuesAtStep
 s() %7B%0A
+%09%09%09
 %09%09re
@@ -2103,70 +2103,199 @@
 urn 
-aggregators;%0A%09%7D%0A%0A%09public long getRuntime() %7B%0A%09%09return runtime;
+(Map%3CString, T%3E) aggregators;%0A%09%09%09%09%7D%0A%09%09%09%7D;%0A%09%09%7D else %7B%0A%09%09%09throw new AggregatorRetrievalException(%22Accumulator results not found.%22,%0A%09%09%09%09%09new RuntimeException(%22Accumulator does not exist.%22));%0A%09%09%7D
 %0A%09%7D%0A

fix FlinkRunnerResult for new PipelineResult