content: /*
   * 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 com.aliyun.odps.examples.graph;
  
  import java.io.IOException;
  import java.util.List;
  
  import org.junit.Assert;
  import org.junit.Test;
  
  import com.aliyun.odps.data.TableInfo;
  import com.aliyun.odps.graph.Combiner;
  import com.aliyun.odps.graph.ComputeContext;
  import com.aliyun.odps.graph.Edge;
  import com.aliyun.odps.graph.GraphJob;
  import com.aliyun.odps.graph.GraphLoader;
  import com.aliyun.odps.graph.MutationContext;
  import com.aliyun.odps.graph.Vertex;
  import com.aliyun.odps.graph.WorkerContext;
  import com.aliyun.odps.io.LongWritable;
  import com.aliyun.odps.io.WritableRecord;
  import com.aliyun.odps.local.common.WareHouse;
  
  public class SSSP {
  
    public static final String START_VERTEX = "sssp.start.vertex.id";
  
    public static class SSSPVertex extends
                                   Vertex<LongWritable, LongWritable, LongWritable, LongWritable> {
  
      private static long startVertexId = -1;
  
      public SSSPVertex() {
        this.setValue(new LongWritable(Long.MAX_VALUE));
      }
  
      public boolean isStartVertex(
          ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context) {
        if (startVertexId == -1) {
          String s = context.getConfiguration().get(START_VERTEX);
          startVertexId = Long.parseLong(s);
        }
        return getId().get() == startVertexId;
      }
  
      @Override
      public void compute(
          ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context,
          Iterable<LongWritable> messages) throws IOException {
        long minDist = isStartVertex(context) ? 0 : Integer.MAX_VALUE;
  
        for (LongWritable msg : messages) {
          if (msg.get() < minDist) {
            minDist = msg.get();
          }
        }
  
        if (minDist < this.getValue().get()) {
          this.setValue(new LongWritable(minDist));
          if (hasEdges()) {
            for (Edge<LongWritable, LongWritable> e : this.getEdges()) {
              context.sendMessage(e.getDestVertexId(), new LongWritable(minDist
                                                                        + e.getValue().get()));
            }
          }
        } else {
          voteToHalt();
        }
      }
  
      @Override
      public void cleanup(
          WorkerContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
          throws IOException {
        context.write(getId(), getValue());
      }
    }
  
    public static class MinLongCombiner extends
                                        Combiner<LongWritable, LongWritable> {
  
      @Override
      public void combine(LongWritable vertexId, LongWritable combinedMessage,
                          LongWritable messageToCombine) throws IOException {
        if (combinedMessage.get() > messageToCombine.get()) {
          combinedMessage.set(messageToCombine.get());
        }
      }
  
    }
  
    public static class SSSPVertexReader extends
                                         GraphLoader<LongWritable, LongWritable, LongWritable, LongWritable> {
  
      @Override
      public void load(
          LongWritable recordNum,
          WritableRecord record,
          MutationContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
          throws IOException {
        SSSPVertex vertex = new SSSPVertex();
        vertex.setId((LongWritable) record.get(0));
        String[] edges = record.get(1).toString().split(";");
        for (int i = 0; i < edges.length; i++) {
          String[] ss = edges[i].split(":");
          vertex.addEdge(new LongWritable(Long.parseLong(ss[0])),
                         new LongWritable(Long.parseLong(ss[1])));
        }
  
        context.addVertexRequest(vertex);
      }
  
    }
  
    public static void main(String[] args) throws Exception {
  
      if (args.length < 2) {
        System.out.println("Usage: <startnode> <input> <output>");
        System.exit(-1);
      }
  
      GraphJob job = new GraphJob();
      job.setGraphLoaderClass(SSSPVertexReader.class);
      job.setVertexClass(SSSPVertex.class);
      job.setCombinerClass(MinLongCombiner.class);
  
      job.set(START_VERTEX, args[0]);
      job.addInput(TableInfo.builder().tableName(args[1]).build());
      job.addOutput(TableInfo.builder().tableName(args[2]).build());
  
      long startTime = System.currentTimeMillis();
      job.run();
      System.out.println("Job Finished in "
                         + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
    }
  
    @Test
    public void test() throws Exception {
      WareHouse wareHouse = WareHouse.getInstance();
      String project = TestUtils.yzy2;
      String outputTable = "sssp_out";
  
      TestUtils.setEnvironment(project);
  
      //If output table exists then delete data (will not delete schema)
      wareHouse.dropTableDataIfExists(project, outputTable, null);
      Assert.assertEquals(true, wareHouse.isTableEmpty(project, outputTable, null));
  
      new SSSP().main(new String[]{"1", "sssp_in", outputTable});
  
      Assert.assertEquals(false, wareHouse.isTableEmpty(project, outputTable, null));
  
      //read output table data
      List<Object[]> result = wareHouse.readData(project, outputTable, null, null, ',');
      Assert.assertEquals(5, result.size());
  
      // Sampling inspection
      Object[] record = result.get(2);
      Assert.assertEquals(2, record.length);
      Assert.assertEquals(true, record[0] instanceof Long);
      Assert.assertEquals(true, record[1] instanceof Long);
      Assert.assertEquals(3L, record[0]);
      Assert.assertEquals(1L, record[1]);
  
    }
  }
  