package com.tu.bdap.sssp;

import java.awt.print.Printable;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
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.graph.Edge;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.pregel.ComputeFunction;
import org.apache.flink.graph.pregel.MessageCombiner;
import org.apache.flink.graph.pregel.MessageIterator;
import org.apache.flink.util.Collector;

public class SSSP_Algorithm {

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

		// Create execution environment
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

		DataSet<Edge<Integer, Double>> edges = env.readTextFile("/home/johannes/Downloads/test.txt")
				.filter(new FilterFunction<String>() {
					@Override
					public boolean filter(String value) throws Exception {
						return value.startsWith("a");
					}
				}).flatMap(new FlatMapFunction<String, Edge<Integer, Double>>() {
					@Override
					public void flatMap(String value, Collector<Edge<Integer, Double>> out) throws Exception {
						String[] values = value.split(" ");
						out.collect(new Edge<Integer, Double>(Integer.parseInt(values[1]), Integer.parseInt(values[2]),
								Double.parseDouble(values[3])));
					}
				});

		Graph<Integer, Double, Double> graph = Graph.fromDataSet(edges, new MapFunction<Integer, Double>() {
			@Override
			public Double map(Integer value) throws Exception {
				return (double) value;
			}
		}, env).getUndirected();
		
		final long srcVertexId = 1;

		Graph<Integer, Double, Double> result = graph.runVertexCentricIteration(new SSSPComputeFunction(srcVertexId),
				new SSSPCombiner(), 20);

		// Extract the vertices as the result
		DataSet<Vertex<Integer, Double>> singleSourceShortestPaths = result.getVertices();

		singleSourceShortestPaths.print();

	}

	@SuppressWarnings("serial")
	public static final class SSSPComputeFunction extends ComputeFunction<Integer, Double, Double, Double> {

		private final long srcId;

		public SSSPComputeFunction(long src) {
			this.srcId = src;
		}

		public void compute(Vertex<Integer, Double> vertex, MessageIterator<Double> messages) {

			double minDistance = (vertex.getId().equals(srcId)) ? 0d : Double.POSITIVE_INFINITY;

			for (Double msg : messages) {
				minDistance = Math.min(minDistance, msg);
			}

			if (minDistance < vertex.getValue()) {
				setNewVertexValue(minDistance);
				for (Edge<Integer, Double> e : getEdges()) {
					sendMessageTo(e.getTarget(), minDistance + e.getValue());
				}
			}
		}
	}

	/**
	 * The messages combiner. Out of all messages destined to a target vertex,
	 * only the minimum distance is propagated.
	 */
	@SuppressWarnings("serial")
	public static final class SSSPCombiner extends MessageCombiner<Integer, Double> {

		public void combineMessages(MessageIterator<Double> messages) {

			double minMessage = Double.POSITIVE_INFINITY;
			for (Double msg : messages) {
				minMessage = Math.min(minMessage, msg);
			}
			sendCombinedMessage(minMessage);
		}
	}

}
@@ -884,31 +884,28 @@
 ataSet%3CEdge%3C
-Integer
+Long
 , Double%3E%3E e
@@ -1176,31 +1176,28 @@
 tring, Edge%3C
-Integer
+Long
 , Double%3E%3E()
@@ -1266,31 +1266,28 @@
 lector%3CEdge%3C
-Integer
+Long
 , Double%3E%3E o
@@ -1376,31 +1376,28 @@
 ct(new Edge%3C
-Integer
+Long
 , Double%3E(In
@@ -1394,32 +1394,30 @@
 Double%3E(
-Integer
+Long
 .parse
-Int
+Long
 (values%5B
@@ -1425,24 +1425,22 @@
 %5D), 
-Integer
+Long
 .parse
-Int
+Long
 (val
@@ -1510,39 +1510,36 @@
 %09%7D);%0D%0A%0D%0A%09%09Graph%3C
-Integer
+Long
 , Double, Double
@@ -1581,39 +1581,36 @@
 new MapFunction%3C
-Integer
+Long
 , Double%3E() %7B%0D%0A%09
@@ -1643,23 +1643,20 @@
 ble map(
-Integer
+Long
  value) 
@@ -1690,22 +1690,32 @@
 urn 
-(d
+D
 ouble
-) value
+.POSITIVE_INFINITY
 ;%0D%0A%09
@@ -1732,24 +1732,8 @@
 env)
-.getUndirected()
 ;%0D%0A%09
@@ -1776,23 +1776,20 @@
 %09%09Graph%3C
-Integer
+Long
 , Double
@@ -1839,16 +1839,22 @@
 eration(
+%0D%0A%09%09%09%09
 new SSSP
@@ -1882,22 +1882,17 @@
 rtexId),
-%0D%0A%09%09%09%09
+ 
 new SSSP
@@ -1906,15 +1906,29 @@
 r(),
- 20);%0D%0A
+%0D%0AInteger.MAX_VALUE);
 %0D%0A%09%09
@@ -1979,31 +1979,28 @@
 aSet%3CVertex%3C
-Integer
+Long
 , Double%3E%3E s
@@ -2198,23 +2198,20 @@
 unction%3C
-Integer
+Long
 , Double
@@ -2364,23 +2364,20 @@
 (Vertex%3C
-Integer
+Long
 , Double
@@ -2699,23 +2699,20 @@
 r (Edge%3C
-Integer
+Long
 , Double
@@ -2714,17 +2714,16 @@
 ouble%3E e
- 
 : getEdg
@@ -2819,16 +2819,18 @@
 %0D%0A%09%7D%0D%0A%0D%0A
+%0D%0A
 %09/**%0D%0A%09 
@@ -3050,23 +3050,20 @@
 ombiner%3C
-Integer
+Long
 , Double

Fix SSSP Flink