//to run in eclipse, add jars below:
//- all jars in $HADOOP_HOME
//- all jars in $HADOOP_HOME/lib

//USAGE:
//- input is the output of DegreeSort
//- output file is then used as batch BFS query file for WorkerOL_auto
//degree threshold -> k

import java.io.*;
import java.util.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

class LineReader {
	private int fileNum;
	private int curfile;
	private String path;
	private String prefix = null;
	private FileSystem fs;
	private FSDataInputStream in = null;

	LineReader(String path) throws IOException {
		this.path = path;
		initialize();
	}

	String fileName(int num) {
		return prefix + "part_" + num;
	}

	private void initialize() throws IOException {

		Configuration conf = new Configuration();
		conf.set("fs.default.name", "hdfs://master:9000");
		fs = FileSystem.get(conf);
		FileStatus[] status = fs.listStatus(new Path(path));
		this.fileNum = status.length;
		this.curfile = 0;
		

		Path filePath = status[0].getPath();
		String path = filePath.toString();
		int slashPos = path.lastIndexOf('/');
		prefix = path.substring(0, slashPos + 1);
		
		in = fs.open(new Path(fileName(this.curfile)));
		
	}

	public String getLine() throws IOException {
		if(in == null) return null;
			
		String line = in.readLine();
		if (line == null) {
			in.close();
			int nextfile = curfile + 1;
			if (nextfile < fileNum) {
				curfile = nextfile;
				in = fs.open(new Path(fileName(this.curfile)));
				line = in.readLine();
			} else {
				in = null;
				return null;
			}
		}
		return line;
	}
}

class TopSelect {

	public static int getDegree(String line)
	{
		StringTokenizer tk = new StringTokenizer(line);
		tk.nextToken();
		return Integer.parseInt(tk.nextToken());
	}
	
	// public static int deg_th=2869;//the deg_th returned from TopSelect_1
	public static int k = 100;
	public static String outfile = "top_vertices.txt";
	public static String infile="hdfs://master:9000/sort/";
	public static int mode = 0; // round_down=0, round_up=1

	public static void main(String[] args) throws IOException {
		
		//k = Integer.parseInt(args[0]);
		//mode =  Integer.parseInt(args[1]);
		//infile =  args[2];
		//outfile =  args[3];
		LineReader reader = new LineReader(infile);
		String line = null;
		Vector<String> result = new Vector<String>();
		while (result.size() < k && (line = reader.getLine()) != null) {
			result.add(line);
		}
		if (mode == 0) {
			line = reader.getLine();
			if (line != null)
			{
				int deg = getDegree(line);
				while(result.size() > 0 && getDegree(result.lastElement()) == deg)
				{
					result.removeElementAt(result.size() - 1);
				}
			}
		}
		else
		{
			while( (line = reader.getLine()) != null && getDegree(line) == getDegree(result.lastElement())  )
			{
				result.add(line);
			}
		}
		FileWriter out = new FileWriter(outfile);
		System.out.println("|H| = " + result.size() );
		for(String res : result)
		{
			out.write(res + "\n");
		}
		out.close();

	}
}