package org.apache.flink.graph.partition.centric;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.graph.Edge;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.GraphAlgorithm;
import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.utils.NullValueEdgeMapper;
import org.apache.flink.types.NullValue;


/**
 * Single Source Shortest Path algorithm implementation using partition centric iterations
 *
 * @param <K>
 * @param <EV>
 */
public class PCSingleSourceShortestPath<K, EV> implements GraphAlgorithm<K, Long, EV, DataSet<Vertex<K, Long>>> {

    public static final String MESSAGE_SENT_CTR = "long:message_sent";
    public static final String MESSAGE_SENT_ITER_CTR = "histogram:message_sent_iter_ctr";
    public static final String ACTIVE_VER_ITER_CTR = "histogram:active_ver_iter_ctr";
    public static final String ITER_CTR = "long:iteration_counter";

    private K srcVertexId;
    private int maxIterations;
    private final PartitionCentricConfiguration configuration;

    /**
     * Single Source Shortest Path algorithm single argument constructor
     *
     * @param srcVertexId ID of the source vertex
     * @param maxIterations Maximum number of iterations of algorithm
     */
    public PCSingleSourceShortestPath(K srcVertexId, int maxIterations) {
        this.srcVertexId = srcVertexId;
        this.maxIterations = maxIterations;
        this.configuration = null;
    }

    /**
     * Single Source Shortest Path algorithm constructor
     *
     * @param srcVertexId ID of the source vertex
     * @param maxIterations Maximum number of iterations of algorithm
     * @param configuration PartitionCentricConfiguration object for iterations
     */
    public PCSingleSourceShortestPath(K srcVertexId, int maxIterations, PartitionCentricConfiguration configuration) {
        this.srcVertexId = srcVertexId;
        this.maxIterations = maxIterations;
        this.configuration = configuration;
    }

    /**
     * Run function for the partition centric SSSP algorithm
     *
     * @param graph The input graph
     * @return The updated graph vertices
     * @throws Exception
     */
    @Override
    public DataSet<Vertex<K, Long>> run(Graph<K, Long, EV> graph) throws Exception {
        PCGraph<K, Long, EV> pcGraph = new PCGraph<>(graph);

        Graph<K, Long, EV> result =
                pcGraph.runPartitionCentricIteration(
                        new SSSPPartitionProcessFunction<K, EV>(),
                        new SSSPVertexUpdateFunction<K, EV>(),
                        configuration, maxIterations);

        return result.getVertices();
    }

    /**
     * User Defined Partition update function
     *
     * @param <K> Type of Vertex ID
     * @param <EV> Type of Edge Value
     */
    public static final class SSSPPartitionProcessFunction<K, EV>  extends PartitionProcessFunction<K, Long, Long, EV> {


        @Override
        public void processPartition(Iterable<Tuple2<Long, Edge<K, EV>>> vertices) throws Exception {

        }
    }

    /**
     * User Defined Partition messaging function
     *
     * @param <K> Type of vertex ID
     * @param <EV> Type of Edge Value
     */
    public static class SSSPVertexUpdateFunction<K, EV> extends VertexUpdateFunction<K, Long, Long, EV> {

        @Override
        public void updateVertex(Iterable<Tuple2<K, Long>> message) {

        }
    }



}


@@ -44,16 +44,75 @@
 ntric;%0A%0A
+import org.apache.flink.api.common.accumulators.Histogram;%0A
 import o
@@ -447,16 +447,73 @@
 lValue;%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A
 %0A%0A/**%0A *
@@ -621,23 +621,60 @@
  %3CK%3E
-%0A * @param %3CEV%3E
+ Type of Vertex ID%0A * @param %3CEV%3E Type of Edge value
 %0A */
@@ -3110,16 +3110,176 @@
  EV%3E %7B%0A%0A
+        private static final long serialVersionUID = 1L;%0A        private static final Logger LOG = LoggerFactory.getLogger(SSSPPartitionProcessFunction.class);%0A
 %0A       
@@ -3388,24 +3388,105 @@
 xception %7B%0A%0A
+            for (Tuple2%3CLong, Edge%3CK, EV%3E%3E vertice : vertices) %7B%0A%0A            %7D%0A%0A
         %7D%0A  
@@ -3835,27 +3835,771 @@
 ) %7B%0A
-%0A        %7D%0A    %7D%0A%0A%0A
+            Histogram vertexHistogram = context.getHistogram(ACTIVE_VER_ITER_CTR);%0A            Long minValue = vertex.getValue();%0A%0A            //find the new minimal value (from current value and message values)%0A            for (Tuple2%3CK, Long%3E m : message) %7B%0A                if (minValue %3E m.f1)%0A                    minValue = m.f1;%0A            %7D%0A%0A            //If vertex value bigger than minValue, update vertex value%0A            //and increase the Active Vertex counter for the next iteration%0A            if (minValue %3C vertex.getValue()) %7B%0A                vertex.setValue(minValue);%0A%0A                if (vertexHistogram != null) %7B%0A                    vertexHistogram.add(context.getSuperstepNumber() + 1);%0A                %7D%0A            %7D%0A        %7D%0A    %7D%0A    
 %0A%7D%0A%0A

Add Vertex Update function