package graphdistance;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Vertex;
import org.apache.flink.types.NullValue;

import java.util.ArrayList;
import java.util.List;

public class APSPData {

	public static final int NUM_VERTICES = 5;

	public static final int SRC_VERTEX_ID = 0;

	//public static final String VERTICES = "0,0\n" + "1,1\n" + "2,2n" + "3,3\n" + "4,4";

	public static DataSet<Vertex<Integer, Tuple2<Integer[], String> >> getDefaultVertexDataSet(ExecutionEnvironment env) {

		List<Vertex<Integer, Tuple2<Integer[], String> >> vertices = new ArrayList<Vertex<Integer, Tuple2<Integer[], String> >>();
		
		// distance vector: distances from all others to current vertex
		Integer[] vertexDistanceVector = new Integer[5];
		for (int i = 0; i< vertexDistanceVector.length; i++) vertexDistanceVector[i]=666;
		
		vertices.add(new Vertex<Integer, Tuple2<Integer[], String>>(0, new Tuple2<Integer[], String>(vertexDistanceVector,"null")));
		vertices.add(new Vertex<Integer, Tuple2<Integer[], String>>(1, new Tuple2<Integer[], String>(vertexDistanceVector,"eins")));
		vertices.add(new Vertex<Integer, Tuple2<Integer[], String>>(2, new Tuple2<Integer[], String>(vertexDistanceVector,"zwei")));
		vertices.add(new Vertex<Integer, Tuple2<Integer[], String>>(3, new Tuple2<Integer[], String>(vertexDistanceVector,"drei")));
		vertices.add(new Vertex<Integer, Tuple2<Integer[], String>>(4, new Tuple2<Integer[], String>(vertexDistanceVector,"vier")));
		
		return env.fromCollection(vertices);
	}

	//public static final String EDGES = "0,1\n" + "0,2\n" + "1,2\n" + "2,3\n" + "2,4\n" + "3,4\n" + "4,0";

	public static final DataSource<Edge<Integer, NullValue>> getDefaultEdgeDataSet(ExecutionEnvironment env) {

		List<Edge<Integer, NullValue>> edges = new ArrayList<Edge<Integer, NullValue>>();
		edges.add(new Edge<Integer, NullValue>(0, 1, NullValue.getInstance()));
		edges.add(new Edge<Integer, NullValue>(0, 2, NullValue.getInstance()));
		edges.add(new Edge<Integer, NullValue>(1, 2, NullValue.getInstance()));
		edges.add(new Edge<Integer, NullValue>(2, 3, NullValue.getInstance()));
		edges.add(new Edge<Integer, NullValue>(2, 4, NullValue.getInstance()));
		edges.add(new Edge<Integer, NullValue>(3, 4, NullValue.getInstance()));
		edges.add(new Edge<Integer, NullValue>(4, 0, NullValue.getInstance()));

		return env.fromCollection(edges);
	}

	//public static final String RESULTED_SINGLE_SOURCE_SHORTEST_PATHS =  "1,0\n" + "2,1\n" + "3,1\n" + "4,2\n" + "5,2";

	private APSPData() {}
}