/*
    * 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.io.formats;
  
  import org.apache.giraph.edge.Edge;
  import org.apache.giraph.graph.Vertex;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.NullWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.hadoop.mapreduce.JobContext;
  import org.apache.hadoop.mapreduce.JobStatus;
  import org.apache.hadoop.mapreduce.OutputCommitter;
  import org.apache.hadoop.mapreduce.TaskAttemptContext;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  
  import java.io.IOException;
  
  /**
   * Writes graph to a dot file (graphviz format). This writes it out in parts. At
   * the end of the job you can use the following to get a single graphviz file:
   *
   * hadoop fs -getmerge /hadoop/output/path data.txt
   */
  public class GraphvizOutputFormat extends TextVertexOutputFormat<
      WritableComparable, Writable, Writable> {
    /** Color of node text */
    private static final String NODE_TEXT_COLOR = "blue:orange";
  
    @Override
    public TextVertexWriter createVertexWriter(TaskAttemptContext context)
      throws IOException, InterruptedException {
      return new VertexWriter();
    }
  
    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext context)
      throws IOException, InterruptedException {
      return new GraphvizOutputCommitter(super.getOutputCommitter(context));
    }
  
    /**
     * Get output directory job is using
     * @param context job context
     * @return Path for output directory
     */
  
    private static Path getOutputDir(JobContext context) {
      return FileOutputFormat.getOutputPath(context);
    }
  
    /**
     * Get path which sorts at the beginning of a directory
     * @param context job context
     * @return Path at beginning
     */
  
    private static Path getPathAtBeginning(JobContext context) {
      return new Path(getOutputDir(context), "____" + System.currentTimeMillis());
    }
  
    /**
     * Get path which sorts at the end of a directory
     * @param context job context
     * @return Path at end
     */
    private static Path getPathAtEnd(JobContext context) {
      return new Path(getOutputDir(context), "zzz_" + System.currentTimeMillis());
    }
  
    /**
     * Write start of graph data
     * @param context task attempt context
     * @throws IOException I/O errors
     */
    private static void writeStart(JobContext context) throws IOException {
      Path path = getPathAtBeginning(context);
      FileSystem fs = path.getFileSystem(context.getConfiguration());
      FSDataOutputStream file = fs.create(path, false);
      file.writeBytes("digraph g {\n");
      file.close();
   }
 
   /**
    * Write end of graph data
    * @param context job attempt context
    * @throws IOException I/O errors
    */
   private static void writeEnd(JobContext context) throws IOException {
     Path path = getPathAtEnd(context);
     FileSystem fs = path.getFileSystem(context.getConfiguration());
     FSDataOutputStream file = fs.create(path, false);
     file.writeBytes("}\n");
     file.close();
   }
 
   /**
    * Add node information to output
    * @param vertex node
    * @param sb string builder
    */
   private static void addNodeInfo(
       Vertex<WritableComparable, Writable, Writable> vertex, StringBuilder sb) {
     sb.append('"').append(vertex.getId()).append('"');
     sb.append(" [").append("label=").append('"').append("<id> ");
     sb.append(vertex.getId());
     if (!(vertex.getValue() instanceof NullWritable)) {
       sb.append("|").append(vertex.getValue());
     }
     sb.append('"').append(",shape=record,fillcolor=")
         .append('"').append(NODE_TEXT_COLOR).append('"')
         .append("];");
   }
 
   /**
    * Write an edge
    * @param sb string builder
    * @param sourceID source vertex ID
    * @param edge the edge
    */
   private static void addEdge(StringBuilder sb, Writable sourceID,
       Edge<WritableComparable, Writable> edge) {
     sb.append(sourceID).append(":id")
         .append(" -> ")
         .append(edge.getTargetVertexId()).append(":id");
     addEdgeInfo(sb, edge);
     sb.append("\n");
   }
 
   /**
    * Add edge information to output
    * @param sb string builder
    * @param edge the edge
    */
   private static void addEdgeInfo(StringBuilder sb,
     Edge<WritableComparable, Writable> edge) {
     if (!(edge.getValue() instanceof NullWritable)) {
       sb.append(" [label=").append(edge.getValue()).append(" ];");
     }
   }
 
   /**
    * Wrapper around output committer which writes our begin/end files.
    */
   private static class GraphvizOutputCommitter extends OutputCommitter {
     /** delegate committer */
     private final OutputCommitter delegate;
 
     /**
      * Constructor with delegate
      * @param delegate committer to use
      */
     private GraphvizOutputCommitter(OutputCommitter delegate) {
       this.delegate = delegate;
     }
 
     @Override public boolean equals(Object o) {
       return delegate.equals(o);
     }
 
     @Override public String toString() {
       return delegate.toString();
     }
 
     @Override public int hashCode() {
       return delegate.hashCode();
     }
 
     @Override public void abortJob(JobContext jobContext, JobStatus.State state)
       throws IOException {
       delegate.abortJob(jobContext, state);
     }
 
     @Override public void abortTask(TaskAttemptContext taskContext)
       throws IOException {
       delegate.abortTask(taskContext);
     }
 
     @Override @Deprecated public void cleanupJob(JobContext context)
       throws IOException {
       delegate.cleanupJob(context);
     }
 
     @Override public void commitJob(JobContext jobContext) throws IOException {
       writeEnd(jobContext);
       delegate.commitJob(jobContext);
     }
 
     @Override public void commitTask(TaskAttemptContext taskContext)
       throws IOException {
       delegate.commitTask(taskContext);
     }
 
     @Override public boolean needsTaskCommit(TaskAttemptContext taskContext)
       throws IOException {
       return delegate.needsTaskCommit(taskContext);
     }
 
     @Override public void setupJob(JobContext jobContext) throws IOException {
       delegate.setupJob(jobContext);
       writeStart(jobContext);
     }
 
     @Override public void setupTask(TaskAttemptContext taskContext)
       throws IOException {
       delegate.setupTask(taskContext);
     }
   }
 
   /**
    * Writes vertices to graphviz files.
    */
   private class VertexWriter extends TextVertexWriter {
     @Override
     public void writeVertex(
       Vertex<WritableComparable, Writable, Writable> vertex)
       throws IOException, InterruptedException {
       StringBuilder sb = new StringBuilder(vertex.getNumEdges() * 10);
       for (Edge<WritableComparable, Writable> edge : vertex.getEdges()) {
         addEdge(sb, vertex.getId(), edge);
       }
       addNodeInfo(vertex, sb);
       getRecordWriter().write(new Text(sb.toString()), null);
     }
   }
 }