/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you 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 org.apache.giraph.examples;
  
  import org.apache.giraph.graph.BasicComputation;
  import org.apache.giraph.edge.Edge;
  import org.apache.giraph.utils.ArrayListWritable;
  import org.apache.giraph.graph.Vertex;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.NullWritable;
  
  import com.google.common.base.Objects;
  import com.google.common.collect.Maps;
  import com.google.common.collect.Sets;
  
  import java.io.IOException;
  import java.util.Map;
  import java.util.Set;
  
  /**
   * Demonstrates triangle closing in simple,
   * unweighted graphs for Giraph.
   *
   * Triangle Closing: Vertex A and B maintain out-edges to C and D
   * The algorithm, when finished, populates all vertices' value with an
   * array of Writables representing all the vertices that each
   * should form an out-edge to (connect with, if this is a social
   * graph.)
   * In this example, vertices A and B would hold empty arrays
   * since they are already connected with C and D. Results:
   * If the graph is undirected, C would hold value, D and D would
   * hold value C, since both are neighbors of A and B and yet both
   * were not previously connected to each other.
   *
   * In a social graph, the result values for vertex X would represent people
   * that are likely a part of a person X's social circle (they know one or more
   * people X is connected to already) but X had not previously met them yet.
   * Given this new information, X can decide to connect to vertices (peoople) in
   * the result array or not.
   *
   * Results at each vertex are ordered in terms of the # of neighbors
   * who are connected to each vertex listed in the final vertex value.
   * The more of a vertex's neighbors who "know" someone, the stronger
   * your social relationship is presumed to be to that vertex (assuming
   * a social graph) and the more likely you should connect with them.
   *
   * In this implementation, Edge Values are not used, but could be
   * adapted to represent additional qualities that could affect the
   * ordering of the final result array.
   */
  public class SimpleTriangleClosingComputation extends BasicComputation<
    IntWritable, SimpleTriangleClosingComputation.IntArrayListWritable,
    NullWritable, IntWritable> {
    /** Vertices to close the triangle, ranked by frequency of in-msgs */
    private Map<IntWritable, Integer> closeMap =
      Maps.<IntWritable, Integer>newHashMap();
  
    @Override
    public void compute(
        Vertex<IntWritable, IntArrayListWritable, NullWritable> vertex,
        Iterable<IntWritable> messages) throws IOException {
      if (getSuperstep() == 0) {
        // send list of this vertex's neighbors to all neighbors
        for (Edge<IntWritable, NullWritable> edge : vertex.getEdges()) {
          sendMessageToAllEdges(vertex, edge.getTargetVertexId());
        }
      } else {
        for (IntWritable message : messages) {
          final int current = (closeMap.get(message) == null) ?
            0 : closeMap.get(message) + 1;
          closeMap.put(message, current);
        }
        // make sure the result values are sorted and
        // packaged in an IntArrayListWritable for output
        Set<Pair> sortedResults = Sets.<Pair>newTreeSet();
        for (Map.Entry<IntWritable, Integer> entry : closeMap.entrySet()) {
          sortedResults.add(new Pair(entry.getKey(), entry.getValue()));
        }
        IntArrayListWritable
          outputList = new IntArrayListWritable();
        for (Pair pair : sortedResults) {
          if (pair.value > 0) {
            outputList.add(pair.key);
         } else {
           break;
         }
       }
       vertex.setValue(outputList);
     }
     vertex.voteToHalt();
   }
 
   /** Quick, immutable K,V storage for sorting in tree set */
   public static class Pair implements Comparable<Pair> {
     /** key
      * @param key the IntWritable key */
     private final IntWritable key;
     /** value
      * @param value the Integer value */
     private final Integer value;
     /** Constructor
      * @param k the key
      * @param v the value
      */
     public Pair(IntWritable k, Integer v) {
       key = k;
       value = v;
     }
     /** key getter
      * @return the key */
     public IntWritable getKey() { return key; }
     /** value getter
      * @return the value */
     public Integer getValue() { return value; }
     /** Comparator to quickly sort by values
      * @param other the Pair to compare with THIS
      * @return the comparison value as an integer */
     @Override
     public int compareTo(Pair other) {
       return other.value - this.value;
     }
 
     @Override
     public boolean equals(Object obj) {
       if (this == obj) {
         return true;
       }
       if (obj instanceof Pair) {
         Pair other = (Pair) obj;
         return Objects.equal(value, other.value);
       }
       return false;
     }
 
     @Override
     public int hashCode() {
       return Objects.hashCode(value);
     }
   }
 
   /** Utility class for delivering the array of vertices THIS vertex
     * should connect with to close triangles with neighbors */
   public static class IntArrayListWritable
     extends ArrayListWritable<IntWritable> {
     /** Default constructor for reflection */
     public IntArrayListWritable() {
       super();
     }
     /** Set storage type for this ArrayListWritable */
     @Override
     @SuppressWarnings("unchecked")
     public void setClass() {
       setClass(IntWritable.class);
     }
   }
 }