/*
 * 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.flink.graph.partition.centric.performance;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.partition.centric.utils.GraphCCRunner;
import org.apache.flink.graph.partition.centric.utils.Telemetry;
import org.apache.flink.types.NullValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Testing the PCConnectedComponents on Twitter Munmun dataset
 */
public class WebGoogle {
    public static void main(String[] args) throws Exception {

        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
        environment.getConfig().disableSysoutLogging();

        Graph<Long, Long, NullValue> graph = Graph
                .fromCsvReader(
                        "data/web-Google/web-Google.data",
                        new MapFunction<Object, Object>() {
                            @Override
                            public Object map(Object value) throws Exception {
                                return value;
                            }
                        }, environment)
                .fieldDelimiterEdges("\t")
                .lineDelimiterEdges("\n")
                .ignoreCommentsEdges("#")
                .vertexTypes(Long.class, Long.class);

        if (args.length < 1) {
            printErr();
        } else if (args[0].equals("1")) {
            GraphCCRunner.detectComponentPC(environment, graph, "out/pcgoogle");
        } else if (args[0].equals("2")) {
            GraphCCRunner.detectComponentVC(environment, graph, "out/vcgoogle");
        } else {
            printErr();
        }
    }

    private static void printErr() {
        System.err.println("Please choose benchmark to run.");
        System.err.printf("Run \"java %s 1\" for partition-centric%n", WebGoogle.class);
        System.err.printf("Run \"java %s 2\" for vertex-centric%n", WebGoogle.class);
        System.exit(-1);
    }
}

@@ -1015,16 +1015,90 @@
 .Graph;%0A
+import org.apache.flink.graph.partition.centric.utils.EnvironmentWrapper;%0A
 import o
@@ -1489,25 +1489,16 @@
         
-Execution
 Environm
@@ -1504,80 +1504,392 @@
 ment
- environment = ExecutionEnvironment.getExecutionEnvironment();%0A        e
+Wrapper wrapper;%0A        if (args.length %3C 2) %7B%0A            printErr();%0A            return;%0A        %7D else if (args%5B1%5D.equals(%22remote%22)) %7B%0A            wrapper = EnvironmentWrapper.newRemote();%0A        %7D else if (args%5B1%5D.equals(%22local%22)) %7B%0A            wrapper = EnvironmentWrapper.newLocal();%0A        %7D else %7B%0A            printErr();%0A            return;%0A        %7D%0A%0A        wrapper.getE
 nvir
@@ -1894,16 +1894,18 @@
 ironment
+()
 .getConf
@@ -2045,14 +2045,34 @@
     
-%22data/
+wrapper.getInputRoot() + %22
 web-
@@ -2375,18 +2375,53 @@
       %7D,
- e
+%0A                        wrapper.getE
 nvironme
@@ -2422,16 +2422,18 @@
 ironment
+()
 )%0A      
@@ -2620,166 +2620,231 @@
     
-if (args.length %3C 1) %7B%0A            printErr();%0A        %7D else if (args%5B0%5D.equals(%221%22)) %7B%0A            GraphCCRunner.detectComponentPC(environment, graph, %22out/
+switch (args%5B0%5D) %7B%0A            case %22pc%22:%0A                GraphCCRunner.detectComponentPC(%0A                        wrapper.getEnvironment(),%0A                        graph,%0A                        wrapper.getOutputRoot() + %22
 pcgo
@@ -2863,42 +2863,50 @@
     
+       
-%7D else if (args%5B0%5D.equals(%222%22)) %7B%0A
+ break;%0A            case %22vc%22:%0A    
     
@@ -2949,62 +2949,201 @@
 tVC(
-environment, graph, %22out/vcgoogle%22);%0A        %7D else %7B%0A
+%0A                        wrapper.getEnvironment(),%0A                        graph,%0A                        wrapper.getOutputRoot() + %22vcgoogle%22);%0A                break;%0A            default:%0A    
     
@@ -3154,32 +3154,55 @@
     printErr();%0A
+                break;%0A
         %7D%0A    %7D%0A
@@ -3347,15 +3347,28 @@
  %25s 
-1%5C%22 for
+pc local%5C%22 for local
  par
@@ -3449,22 +3449,243 @@
  %25s 
-2%5C%22 for vertex
+vc local%5C%22 for local vertex-centric%25n%22, WebGoogle.class);%0A        System.err.printf(%22Run %5C%22java %25s pc remote%5C%22 for remote partition-centric%25n%22, WebGoogle.class);%0A        System.err.printf(%22Run %5C%22java %25s pc remote%5C%22 for remote partition
 -cen

Enable WebGoogle to run remotely