// Path: src/main/java/org/apache/jena/grande/NodeEncoder.java
// public class NodeEncoder {
// 
// 	private static final PrefixMap emptyPrefixMap = new PrefixMap();
// 	
// 	public static String asString(Node node) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, emptyPrefixMap);
//         return sw.toString() ; 
// 	}
// 	
// 	public static String asString(Node node, PrefixMap prefixMap) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, prefixMap);
//         return sw.toString() ; 
// 	}
// 
// 	public static Node asNode(String str) {
// 		return NodeFactory.parseNode(str);
// 	}
// 
// }
// 
// Path: src/main/java/org/apache/jena/grande/mapreduce/io/NodeWritable.java
// public class NodeWritable extends BinaryComparable implements WritableComparable<BinaryComparable> {
// 
//     private Node node;
//     private byte[] bytes;
//     private int length;
//     
//     public NodeWritable(){
//     	this( Node.ANY );
//     }
//     
//     public NodeWritable(Node node) {
//         this.node = node;
//         this.bytes = toBytes(node);
//         this.length = bytes.length;
//     }
//     
//     @Override
//     public void readFields(DataInput in) throws IOException {
//         length = WritableUtils.readVInt(in);
//         bytes = new byte[length];
//         in.readFully(bytes, 0, length);
//         node = NodeEncoder.asNode(new String(bytes));
//     }
// 
//     @Override
//     public void write(DataOutput out) throws IOException {
//         WritableUtils.writeVInt(out, length);
//         out.write(bytes, 0, length);
//     }
// 
//     @Override
//     public byte[] getBytes() {
//         return bytes;
//     }
// 
//     @Override
//     public int getLength() {
//         return length;
//     }
//     
//     public Node getNode() {
//         return node;
//     }
//     
//     private byte[] toBytes(Node node) {
//         StringWriter out = new StringWriter();
//         out.getBuffer().length();
//         OutputLangUtils.output(out, node, null, null);
//         try {
//             return out.toString().getBytes("UTF-8");
//         } catch (UnsupportedEncodingException e) {
//             throw new JenaGrandeException(e);
//         }
//     }
// 
//     @Override
//     public String toString() {
//         return node.toString();
//     }
// 
// }


import java.io.IOException;
import java.util.Map;
import org.apache.giraph.graph.BspUtils;
import org.apache.giraph.graph.Vertex;
import org.apache.giraph.io.TextVertexInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.jena.grande.NodeEncoder;
import org.apache.jena.grande.mapreduce.io.NodeWritable;
import org.openjena.riot.Lang;
import org.openjena.riot.RiotLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Maps;
import com.hp.hpl.jena.graph.Graph;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.sparql.vocabulary.FOAF;
import com.hp.hpl.jena.util.iterator.ExtendedIterator;
		log.debug("createVertexReader({}, {}) --> {}", new Object[]{split, context, result});
	    return result;
	}

	public class TurtleVertexReader extends TextVertexReader {

		private final Logger log = LoggerFactory.getLogger(TurtleVertexReader.class);

		@Override
		public boolean nextVertex() throws IOException, InterruptedException {
			boolean result = getRecordReader().nextKeyValue();
			log.debug("nextVertex() --> {}", result);
			return result;
		}

		@Override
		public Vertex<NodeWritable, IntWritable, NodeWritable, IntWritable> getCurrentVertex() throws IOException, InterruptedException {
			Configuration conf = getContext().getConfiguration();
			Vertex<NodeWritable, IntWritable, NodeWritable, IntWritable> vertex = BspUtils.createVertex(conf);
			Text line = getRecordReader().getCurrentValue();
			NodeWritable vertexId = getVertexId(line);
			Graph graph = RiotLoader.graphFromString(line.toString(), Lang.TURTLE, "");
			Map<NodeWritable, NodeWritable> edgeMap = getEdgeMap(vertexId, graph);
			vertex.initialize(vertexId, null, edgeMap);
			log.debug("getCurrentVertex() --> {}", vertex);
			return vertex;
		}
		
		private NodeWritable getVertexId( Text line ) {
			String str = line.toString();
// Path: src/main/java/org/apache/jena/grande/NodeEncoder.java
// public class NodeEncoder {
// 
// 	private static final PrefixMap emptyPrefixMap = new PrefixMap();
// 	
// 	public static String asString(Node node) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, emptyPrefixMap);
//         return sw.toString() ; 
// 	}
// 	
// 	public static String asString(Node node, PrefixMap prefixMap) {
//         StringWriter sw = new StringWriter() ;
//         NodeFmtLib.serialize(sw, node, null, prefixMap);
//         return sw.toString() ; 
// 	}
// 
// 	public static Node asNode(String str) {
// 		return NodeFactory.parseNode(str);
// 	}
// 
// }
// 
// Path: src/main/java/org/apache/jena/grande/mapreduce/io/NodeWritable.java
// public class NodeWritable extends BinaryComparable implements WritableComparable<BinaryComparable> {
// 
//     private Node node;
//     private byte[] bytes;
//     private int length;
//     
//     public NodeWritable(){
//     	this( Node.ANY );
//     }
//     
//     public NodeWritable(Node node) {
//         this.node = node;
//         this.bytes = toBytes(node);
//         this.length = bytes.length;
//     }
//     
//     @Override
//     public void readFields(DataInput in) throws IOException {
//         length = WritableUtils.readVInt(in);
//         bytes = new byte[length];
//         in.readFully(bytes, 0, length);
//         node = NodeEncoder.asNode(new String(bytes));
//     }
// 
//     @Override
//     public void write(DataOutput out) throws IOException {
//         WritableUtils.writeVInt(out, length);
//         out.write(bytes, 0, length);
//     }
// 
//     @Override
//     public byte[] getBytes() {
//         return bytes;
//     }
// 
//     @Override
//     public int getLength() {
//         return length;
//     }
//     
//     public Node getNode() {
//         return node;
//     }
//     
//     private byte[] toBytes(Node node) {
//         StringWriter out = new StringWriter();
//         out.getBuffer().length();
//         OutputLangUtils.output(out, node, null, null);
//         try {
//             return out.toString().getBytes("UTF-8");
//         } catch (UnsupportedEncodingException e) {
//             throw new JenaGrandeException(e);
//         }
//     }
// 
//     @Override
//     public String toString() {
//         return node.toString();
//     }
// 
// }


// Path: src/main/java/org/apache/jena/grande/giraph/TurtleVertexInputFormat.java
import java.io.IOException;
import java.util.Map;
import org.apache.giraph.graph.BspUtils;
import org.apache.giraph.graph.Vertex;
import org.apache.giraph.io.TextVertexInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.jena.grande.NodeEncoder;
import org.apache.jena.grande.mapreduce.io.NodeWritable;
import org.openjena.riot.Lang;
import org.openjena.riot.RiotLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Maps;
import com.hp.hpl.jena.graph.Graph;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.sparql.vocabulary.FOAF;
import com.hp.hpl.jena.util.iterator.ExtendedIterator;

		log.debug("createVertexReader({}, {}) --> {}", new Object[]{split, context, result});
	    return result;
	}

	public class TurtleVertexReader extends TextVertexReader {

		private final Logger log = LoggerFactory.getLogger(TurtleVertexReader.class);

		@Override
		public boolean nextVertex() throws IOException, InterruptedException {
			boolean result = getRecordReader().nextKeyValue();
			log.debug("nextVertex() --> {}", result);
			return result;
		}

		@Override
		public Vertex<NodeWritable, IntWritable, NodeWritable, IntWritable> getCurrentVertex() throws IOException, InterruptedException {
			Configuration conf = getContext().getConfiguration();
			Vertex<NodeWritable, IntWritable, NodeWritable, IntWritable> vertex = BspUtils.createVertex(conf);
			Text line = getRecordReader().getCurrentValue();
			NodeWritable vertexId = getVertexId(line);
			Graph graph = RiotLoader.graphFromString(line.toString(), Lang.TURTLE, "");
			Map<NodeWritable, NodeWritable> edgeMap = getEdgeMap(vertexId, graph);
			vertex.initialize(vertexId, null, edgeMap);
			log.debug("getCurrentVertex() --> {}", vertex);
			return vertex;
		}
		
		private NodeWritable getVertexId( Text line ) {
			String str = line.toString();
			NodeWritable vertexId = new NodeWritable(NodeEncoder.asNode(str.substring(0, str.indexOf(' '))));