/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package hdfsapi;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 *
 * @author sc-md
 */
public class FileSystemOperations
{

    public FileSystemOperations()
    {
    }

    public void mkdir(String dir, Configuration conf) throws IOException
    {
        FileSystem fileSystem = FileSystem.get(conf);

        Path path = new Path(dir);
        if (fileSystem.exists(path))
        {
            System.out.println("Dir " + dir + " already not exists");
            return;
        }

        fileSystem.mkdirs(path);

        fileSystem.close();
    }
    
    public void rmdir(String hadoopDirPath, Configuration conf) throws IOException
    {
        FileSystem fileSystem = FileSystem.get(conf);
        Path path;
        
        path = new Path(hadoopDirPath);
        if (!fileSystem.exists(path))
        {
            System.out.println("File " + hadoopDirPath + " does not exists");
            return;
        }

        fileSystem.delete(new Path(hadoopDirPath), true);  // true=recursive, remove all contents

        fileSystem.close();
    }

    public void addFile(String filePath, String hadoopFilePath, Configuration conf) throws IOException
    {
        FileSystem fileSystem = FileSystem.get(conf);

        // Get the filename out of the file path
//        String destFilePath;
//        String filename;
//        filename = filePath.substring(filePath.lastIndexOf('/') + 1, filePath.length());
//        destFilePath = buildDestFilePath(destDir, filename);
//        System.out.println("Adding file to " + destFilePath);

        // Check if the file already exists in hadoop
        // DON'T DESTROY EXISTING FILE!!
        Path path = new Path(hadoopFilePath);
        if (fileSystem.exists(path))
        {
            System.out.println("File " + hadoopFilePath + " already exists in hadoop");
            return;
        }

        // Create the destination file. 
        // FSDataOutputStream out;
        // out = fileSystem.create(path);
        // copyToHadoopFile(sourceFilePath, out);
        // out.close();

        // Create the destination file. 
        if (fileSystem.createNewFile(path) == true)
        {
            // Copy source file contents to destination file
            copyToHadoopFile(fileSystem, filePath, hadoopFilePath);
        }

        fileSystem.close();
    }

    public void readFile(String hadoopFilePath, String filePath, Configuration conf) throws IOException
    {
        FileSystem fileSystem = FileSystem.get(conf);
        Path path;

        // Check if the file we want to read does exist in hadoop
        path = new Path(hadoopFilePath);
        if (!fileSystem.exists(path))
        {
            System.out.println("File " + filePath + " does not exist in hadoop");
            return;
        }

        copyFromHadoopFile(fileSystem, hadoopFilePath, filePath);
        
        fileSystem.close();
    }
        
    // Identical to rmdir
    public void deleteFile(String hadoopFilePath, Configuration conf) throws IOException
    {
        FileSystem fileSystem = FileSystem.get(conf);
        Path path;
        
        path = new Path(hadoopFilePath);
        if (!fileSystem.exists(path))
        {
            System.out.println("File " + hadoopFilePath + " does not exists");
            return;
        }

        // recursive can be true/false, makes no difference for a file
        fileSystem.delete(new Path(hadoopFilePath), true);  

        fileSystem.close();
    }

    private String buildDestFilePath(String destDirPath, String filename)
    {
        String destFilePath;

        // Check if path ends in '/', add one if it doesn't.
        if (destDirPath.charAt(destDirPath.length() - 1) != '/')
        {
            destFilePath = destDirPath + "/" + filename;
        }
        else
        {
            destFilePath = destDirPath + filename;
        }
        return destFilePath;
    }

//    private void copyToHadoopFile(String localSourceFilePath, FSDataOutputStream out) throws IOException
//    {
//        File sourceFile;
//        FileInputStream sourceFS;
//        InputStream in;
//        byte[] bytes = new byte[1024];      // Read in 1K blocks
//        int numBytes = 0;       
//        
//        sourceFile = new File(localSourceFilePath);
//        sourceFS = new FileInputStream(sourceFile);        
//        in = new BufferedInputStream(sourceFS);
//               
//        while ((numBytes = in.read(bytes)) > 0)
//        {
//            out.write(bytes, 0, numBytes);
//        }
//        
//        in.close();  
//    }
    private void copyToHadoopFile(FileSystem fileSystem, String filePath, String hadoopFilePath) throws IOException
    {
        File sourceFile;
        FileInputStream sourceFS;
        InputStream in;
        FSDataOutputStream out;
        byte[] bytes = new byte[1024];  // Read in 1K blocks
        int numBytes = 0;

        // Read from the local file
        sourceFile = new File(filePath);
        sourceFS = new FileInputStream(sourceFile);
        in = new BufferedInputStream(sourceFS);

        // Write to the Hadoop file
        out = fileSystem.create(new Path(hadoopFilePath), true);  // Overwrite contents

        while ((numBytes = in.read(bytes)) > 0)
        {
            out.write(bytes, 0, numBytes);
        }

        // Close all the file descriptors
        in.close();
        out.close();
    }

    private void copyFromHadoopFile(FileSystem fileSystem, String hadoopFilePath, String filePath) throws IOException
    {
        File file;
        FSDataInputStream in;
        FileOutputStream sourceFS;
        OutputStream out;
        byte[] bytes = new byte[1024];
        int numBytes = 0;

        // Read from the Hadoop file
        in = fileSystem.open(new Path(hadoopFilePath));

        // Write to the local file
        file = new File(filePath);
        sourceFS = new FileOutputStream(file);
        out = new BufferedOutputStream(sourceFS);

        while ((numBytes = in.read(bytes)) > 0)
        {
            out.write(bytes, 0, numBytes);
        }

        in.close();
        out.close();
    }
}