/**
 * Copyright 2015 Delft University of Technology
 *
 * 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 nl.tudelft.graphalytics.giraph.bfs;

import static nl.tudelft.graphalytics.giraph.bfs.BreadthFirstSearchConfiguration.SOURCE_VERTEX;

import java.io.IOException;

import org.apache.giraph.graph.BasicComputation;
import org.apache.giraph.graph.Vertex;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;

/**
 * Implementation of a simple BFS (SSSP) on an unweighted, directed graph.
 * 
 * @author Tim Hegeman
 */
public class BreadthFirstSearchComputation extends BasicComputation<LongWritable, LongWritable, NullWritable, LongWritable> {

	/** Constant vertex value representing an unvisited vertex */ 
	private static final LongWritable UNVISITED = new LongWritable(Long.MAX_VALUE);
	
	@Override
	public void compute(Vertex<LongWritable, LongWritable, NullWritable> vertex,
			Iterable<LongWritable> messages) throws IOException {
		LongWritable bfsDepth = new LongWritable(getSuperstep());
		
		if (getSuperstep() == 0) {
			// During the first superstep only the source vertex should be active
			if (vertex.getId().get() == SOURCE_VERTEX.get(getConf())) {
				vertex.setValue(bfsDepth);
				sendMessageToAllEdges(vertex, vertex.getValue());
			} else {
				vertex.setValue(UNVISITED);
			}
		} else {
			// If this vertex was not yet visited, set the vertex depth and propagate to neighbours
			if (vertex.getValue().get() == UNVISITED.get()) {
				vertex.setValue(bfsDepth);
				sendMessageToAllEdges(vertex, vertex.getValue());
			}
		}
		
		// Always halt so the compute method is only executed for those vertices
		// that have an incoming message
		vertex.voteToHalt();
	}

}

@@ -757,37 +757,204 @@
 EX;%0A
-%0Aimport java.io.IOException;%0A
+import static nl.tudelft.graphalytics.giraph.bfs.BreadthFirstSearchConfiguration.SOURCE_VERTEX_KEY;%0A%0Aimport java.io.IOException;%0A%0Aimport org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
 %0Aimp
@@ -1446,28 +1446,20 @@
 c final 
-L
+l
 ong
-Writable
  UNVISIT
@@ -1463,20 +1463,207 @@
 SITED = 
-new 
+Long.MAX_VALUE;%0A%09/** Source vertex ID read at the start of the algorithm execution */%0A%09private long sourceVertexId = -1L;%0A%0A%09@Override%0A%09public void setConf(ImmutableClassesGiraphConfiguration%3C
 LongWrit
@@ -1670,27 +1670,121 @@
 able
-(
+, 
 Long
-.MAX_VALUE
+Writable, NullWritable%3E conf) %7B%0A%09%09super.setConf(conf);%0A%09%09sourceVertexId = SOURCE_VERTEX.get(getConf()
 );%0A%09
+%7D%0A
 %0A%09@O
@@ -1928,28 +1928,20 @@
 ion %7B%0A%09%09
-L
+l
 ong
-Writable
  bfsDept
@@ -1948,25 +1948,8 @@
 h = 
-new LongWritable(
 getS
@@ -1958,17 +1958,16 @@
 erstep()
-)
 ;%0A%09%09%0A%09%09i
@@ -2100,36 +2100,22 @@
  == 
-SOURCE_VERTEX.get(getConf())
+sourceVertexId
 ) %7B%0A
@@ -2121,33 +2121,33 @@
 %0A%09%09%09%09vertex.
-s
+g
 etValue(
 bfsDepth);%0A%09
@@ -2126,32 +2126,38 @@
 vertex.getValue(
+).set(
 bfsDepth);%0A%09%09%09%09s
@@ -2228,25 +2228,25 @@
 %09vertex.
-s
+g
 etValue(
 UNVISITE
@@ -2237,16 +2237,22 @@
 etValue(
+).set(
 UNVISITE
@@ -2405,22 +2405,16 @@
 NVISITED
-.get()
 ) %7B%0A%09%09%09%09
@@ -2424,17 +2424,17 @@
 tex.
-s
+g
 etValue(
 bfsD
@@ -2429,16 +2429,22 @@
 etValue(
+).set(
 bfsDepth

Optimize BFS code by reducing the number of objects created