/*
 * (c) 2017 Real-Time Technology Solutions, Inc. All Rights Reserved.
 * 
 * THIS SOFTWARE IS PROVIDED 'AS IS' AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED.  IN NO EVENT SHALL REAL-TIME TECHNOLOGY SOLUTIONS, INC.
 * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
 * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
 * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
 * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, 
 * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package com.rttsweb.querysurge;

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class HdfsClient {
	private static final String LINUXSEPARATOR = "/";
	private static boolean verbose = false;

	public static void main(String[] args) throws IOException {
		String server = args[0];
		String hdfsDir = args[1];
		String fileName = args[2];
		String localWritePath = args[3];
		if ((args.length > 4) && (args[4] != null))
			verbose = Boolean.parseBoolean(args[4]);
		hdfsFileDownload(server, hdfsDir, fileName, localWritePath, verbose);
	}
	
	public static void hdfsFileDownload(String hdfsAddress, String hdfsDirPath, String fileName, String targetPath, boolean verbose) throws IOException {
		HdfsClient.verbose = verbose;
		String [] parts = hdfsAddress.split(":");
		String hdfsServer = parts[0];
		String hdfsPort = parts[1];
		hdfsFileDownload(hdfsServer, hdfsPort, hdfsDirPath, fileName, targetPath);
	}
	
	private static void hdfsFileDownload(String hdfsServer, String hdfsPort, String hdfsDirPath, String fileName, String targetPath) throws IOException {
		// handle the hdfs dir path formatting to make sure the path ends with a separator char
		String formattedHdfsDirPath = (hdfsDirPath.endsWith(LINUXSEPARATOR)) ? hdfsDirPath : hdfsDirPath + LINUXSEPARATOR;
		// handle the local OS file system dir path to make sure that the path ends with a separator char
		String formattedTargetDirPath = (targetPath.endsWith(File.separator)) ? targetPath : targetPath + File.separator;
		
		// set up a Configuration pointing to the hdfs server
		Configuration conf = new Configuration ();
		String path = "hdfs://" + hdfsServer + ":" + hdfsPort;
		conf.set("fs.default.name", path);
		if (verbose) {
			System.out.println(path);
		}
		
		// get a ref to the hdfs FileSystem
		FileSystem fs = FileSystem.get(conf);
		// get a FileStatus ref for the specified hdfs file
		FileStatus[] filestatus = fs.listStatus(new Path(formattedHdfsDirPath + fileName));
		
		try {
			// make sure the hdfs file path is a file path and not a dir path
        	if (!filestatus[0].isDirectory()) {
        		// get an hdfs Path ref to the hdfs file...
        		Path hdfsFilepath = filestatus[0].getPath();
        		// ...and copy the file to the local file system
        		fs.copyToLocalFile(false, hdfsFilepath, new Path(formattedTargetDirPath), true);
        		if (verbose) {
        			System.out.println("File: " + hdfsFilepath.getName() + " downloaded");
        		}
        	} else {
        		System.out.println("Directory specified, not File");
        	}
        } catch (Exception e) {
        	if (verbose) {
        		e.printStackTrace();
        	} else {
        		System.out.println(e.toString());
        	}
        } finally {
        	// cleanup
        	fs.close();
        }
    }
}