/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.flink.graph.example;

import org.apache.flink.api.common.ProgramDescription;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.example.utils.SingleSourceShortestPathsDataUnweighted;
import org.apache.flink.graph.library.SingleSourceShortestPathsUnweighted;
import org.apache.flink.types.NullValue;

public class SingleSourceShortestPathsExampleUnweighted implements ProgramDescription {

	public static void main(String[] args) throws Exception {

		if (!parseParameters(args)) {
			return;
		}
		
		//Long tempLong = Long.MAX_VALUE;
		//System.out.println("DEBUG Long.MAX_VALUE: " + tempLong);
		System.out.println("DEBUG srcVertexId: " + srcVertexId);
		

		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

		DataSet<Vertex<Long, Long>> vertices = getVerticesDataSet(env);

		DataSet<Edge<Long, NullValue>> edges = getEdgesDataSet(env);

		Graph<Long, Long, NullValue> graph = Graph.fromDataSet(vertices, edges, env);

		DataSet<Vertex<Long, Long>> singleSourceShortestPathsUnweighted = graph
				.run(new SingleSourceShortestPathsUnweighted<Long>(srcVertexId,
						maxIterations)).getVertices();

		// emit result
		if (fileOutput) {
			singleSourceShortestPathsUnweighted.writeAsCsv(outputPath, "\n", " ");
		} else {
			singleSourceShortestPathsUnweighted.print();
		}

		env.execute("Single Source Shortest Paths Example");
	}

	@Override
	public String getDescription() {
		return "Single Source Shortest Paths";
	}

	// ******************************************************************************************************************
	// UTIL METHODS
	// ******************************************************************************************************************

	private static boolean fileOutput = false;

	private static Long srcVertexId = null;

	private static String verticesInputPath = null;

	private static String edgesInputPath = null;

	private static String outputPath = null;

	private static int maxIterations = 5;

	private static boolean parseParameters(String[] args) {

		if (args.length > 0) {
			if (args.length == 5) {
				fileOutput = true;
				srcVertexId = Long.parseLong(args[0]);
				verticesInputPath = args[1];
				edgesInputPath = args[2];
				outputPath = args[3];
				maxIterations = Integer.parseInt(args[4]);
			} else {
				System.err.println("Usage: SingleSourceShortestPaths <source vertex id>" +
						" <input vertices path> <input edges path> <output path> <num iterations>");
				return false;
			}
		}
		return true;
	}

	@SuppressWarnings("serial")
	private static DataSet<Vertex<Long, Long>> getVerticesDataSet(ExecutionEnvironment env) {
		if (fileOutput) {
			return env.readCsvFile(verticesInputPath)
					.fieldDelimiter(" ")
					.lineDelimiter("\n")
					.types(Long.class, Long.class)
					.map(new MapFunction<Tuple2<Long, Long>, Vertex<Long, Long>>() {

						@Override
						public Vertex<Long, Long> map(Tuple2<Long, Long> tuple2) throws Exception {
							return new Vertex<Long, Long>(tuple2.f0, tuple2.f1);
						}
					});
		} else {
			System.err.println("Usage: SingleSourceShortestPathsUnweighted <source vertex id>" +
					" <input vertices path> <input edges path> <output path> <num iterations>");
			return SingleSourceShortestPathsDataUnweighted.getDefaultVertexDataSet(env);
		}
	}

	@SuppressWarnings("serial")
	private static DataSet<Edge<Long, NullValue>> getEdgesDataSet(ExecutionEnvironment env) {
		if (fileOutput) {
			return env.readCsvFile(edgesInputPath)
					.fieldDelimiter(" ")
					.lineDelimiter("\n")
					.types(Long.class, Long.class)
					.map(new MapFunction<Tuple2<Long,Long>, Edge<Long, NullValue>>() {
						
						@Override
						public Edge<Long, NullValue> map(Tuple2<Long, Long> tuple2) throws Exception {
							return new Edge<Long, NullValue>(tuple2.f0, tuple2.f1, NullValue.getInstance());
						}
					});

		} else {
			System.err.println("Usage: SingleSourceShortestPathsUnweighted <source vertex id>" +
					" <input vertices path> <input edges path> <output path> <num iterations>");
			return SingleSourceShortestPathsDataUnweighted.getDefaultEdgeDataSet(env);
		}
	}
}