[FLINK-3181] [gelly] avoid unnecessary messages in SSSP examples and library method

This closes #1467

flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/SingleSourceShortestPaths.scala

@@ -113,8 +113,10 @@ object SingleSourceShortestPaths {
     MessagingFunction[Long, Double, Double, Double] {
 
     override def sendMessages(vertex: Vertex[Long, Double]) {
-      for (edge: Edge[Long, Double] <- getEdges) {
-        sendMessageTo(edge.getTarget, vertex.getValue + edge.getValue)
+      if (vertex.getValue < Double.PositiveInfinity) {
+        for (edge: Edge[Long, Double] <- getEdges) {
+          sendMessageTo(edge.getTarget, vertex.getValue + edge.getValue)
+        }
       }
     }
   }

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/example/SingleSourceShortestPaths.java

@@ -135,8 +135,10 @@ public class SingleSourceShortestPaths implements ProgramDescription {
 
 		@Override
 		public void sendMessages(Vertex<Long, Double> vertex) {
-			for (Edge<Long, Double> edge : getEdges()) {
-				sendMessageTo(edge.getTarget(), vertex.getValue() + edge.getValue());
+			if (vertex.getValue() < Double.POSITIVE_INFINITY) {
+				for (Edge<Long, Double> edge : getEdges()) {
+					sendMessageTo(edge.getTarget(), vertex.getValue() + edge.getValue());
+				}
 			}
 		}
 	}

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java

@@ -108,10 +108,11 @@ public class SingleSourceShortestPaths<K> implements GraphAlgorithm<K, Double, D
 	public static final class MinDistanceMessenger<K> extends MessagingFunction<K, Double, Double, Double> {
 
 		@Override
-		public void sendMessages(Vertex<K, Double> vertex)
-				throws Exception {
-			for (Edge<K, Double> edge : getEdges()) {
-				sendMessageTo(edge.getTarget(), vertex.getValue() + edge.getValue());
+		public void sendMessages(Vertex<K, Double> vertex) {
+			if (vertex.getValue() < Double.POSITIVE_INFINITY) {
+				for (Edge<K, Double> edge : getEdges()) {
+					sendMessageTo(edge.getTarget(), vertex.getValue() + edge.getValue());
+				}
 			}
 		}
 	}