import org.apache.giraph.graph.BasicComputation;
import org.apache.giraph.edge.Edge;
import org.apache.giraph.graph.Vertex;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;

import java.io.IOException;

/**
 * Implementation of the connected component algorithm that identifies
 * connected components and assigns each vertex its "component
 * identifier" (the smallest vertex id in the component).
 */
public class ConnectedComponentsComputation extends
    BasicComputation<IntWritable, IntWritable, NullWritable, IntWritable> {
  /**
   * Propagates the smallest vertex id to all neighbors. Will always choose to
   * halt and only reactivate if a smaller id has been sent to it.
   *
   * @param vertex Vertex
   * @param messages Iterator of messages from the previous superstep.
   * @throws IOException
   */
  @Override
  public void compute(
      Vertex<IntWritable, IntWritable, NullWritable> vertex,
      Iterable<IntWritable> messages) throws IOException {
      //TODO
  }
}

@@ -22,30 +22,17 @@
 aph.
-graph.BasicComputation
+edge.Edge
 ;%0Aim
@@ -50,33 +50,46 @@
 ache.giraph.
-edge.Edge
+graph.BasicComputation
 ;%0Aimport org
@@ -482,16 +482,20 @@
 extends%0A
+    
     Basi
@@ -562,16 +562,18 @@
 able%3E %7B%0A
+  
   /**%0A  
@@ -573,16 +573,18 @@
  /**%0A   
+ 
+ 
 * Propag
@@ -654,16 +654,18 @@
 e to%0A   
+  
 * halt a
@@ -723,21 +723,25 @@
  it.%0A   
+  
 *%0A   
+ 
+ 
 * @param
@@ -752,15 +752,19 @@
 tex 
+  
 Vertex%0A
+  
    *
@@ -830,16 +830,18 @@
 erstep.%0A
+  
    * @th
@@ -860,19 +860,23 @@
 tion%0A   
+  
 */%0A
+  
   @Overr
@@ -879,16 +879,18 @@
 verride%0A
+  
   public
@@ -903,23 +903,16 @@
 compute(
-%0A      
 Vertex%3CI
@@ -961,81 +961,1642 @@
 tex,
-%0A      Iterable%3CIntWritable%3E messages) throws IOException %7B%0A      //TODO%0A
+ Iterable%3CIntWritable%3E messages) throws IOException %7B%0A        int currentComponent = vertex.getValue().get();%0A// First superstep is special, because we can simply look at the neighbors%0A        if (getSuperstep() == 0) %7B%0A            for (Edge%3CIntWritable, NullWritable%3E edge : vertex.getEdges()) %7B%0A                int neighbor = edge.getTargetVertexId().get();%0A                if (neighbor %3C currentComponent) %7B%0A                    currentComponent = neighbor;%0A                %7D%0A            %7D%0A// Only need to send value if it is not the own id%0A            if (currentComponent != vertex.getValue().get()) %7B%0A                vertex.setValue(new IntWritable(currentComponent));%0A                for (Edge%3CIntWritable, NullWritable%3E edge : vertex.getEdges()) %7B%0A                    IntWritable neighbor = edge.getTargetVertexId();%0A                    if (neighbor.get() %3E currentComponent) %7B%0A                        sendMessage(neighbor, vertex.getValue());%0A                    %7D%0A                %7D%0A            %7D%0A            vertex.voteToHalt();%0A            return;%0A        %7D%0A        boolean changed = false;%0A// did we get a smaller id ?%0A        for (IntWritable message : messages) %7B%0A            int candidateComponent = message.get();%0A            if (candidateComponent %3C currentComponent) %7B%0A                currentComponent = candidateComponent;%0A                changed = true;%0A            %7D%0A        %7D%0A// propagate new component id to the neighbors%0A        if (changed) %7B%0A            vertex.setValue(new IntWritable(currentComponent));%0A            sendMessageToAllEdges(vertex, vertex.getValue());%0A        %7D%0A        vertex.voteToHalt();%0A  
   %7D%0A

add connected components computation