/**
 * 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.ambari.fast_hdfs_resource;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.File;
import java.lang.System;
import java.util.ArrayList;
import java.util.HashSet;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.fs.FileSystem;

/**
 * Used to: 1) copy files/directories from localFS to hadoopFs 2) create empty
 * files/directories in hadoopFs
 */
public class Resource {
  private String source;
  private String target;
  private String type;
  private String action;
  private String owner;
  private String group;
  private String mode;
  private boolean recursiveChown;
  private boolean recursiveChmod;
  private boolean changePermissionforParents;

  public String getSource() {
    return source;
  }

  public void setSource(String source) {
    this.source = source;
  }

  public String getTarget() {
    return target;
  }

  public void setTarget(String target) {
    this.target = target;
  }

  public String getType() {
    return type;
  }

  public void setType(String type) {
    this.type = type;
  }

  public String getAction() {
    return action;
  }

  public void setAction(String action) {
    this.action = action;
  }

  public String getOwner() {
    return owner;
  }

  public void setOwner(String owner) {
    this.owner = owner;
  }

  public String getGroup() {
    return group;
  }

  public void setGroup(String group) {
    this.group = group;
  }

  public String getMode() {
    return mode;
  }

  public void setMode(String mode) {
    this.mode = mode;
  }

  public boolean isRecursiveChown() {
    return recursiveChown;
  }

  public void setRecursiveChown(boolean recursiveChown) {
    this.recursiveChown = recursiveChown;
  }

  public boolean isRecursiveChmod() {
    return recursiveChmod;
  }

  public void setRecursiveChmod(boolean recursiveChmod) {
    this.recursiveChmod = recursiveChmod;
  }
  
  public boolean isChangePermissionOnParents() {
    return changePermissionforParents;
  }

  public void setChangePermissionOnParents(boolean changePermissionforParents) {
    this.changePermissionforParents = changePermissionforParents;
  }

  
  
  
  @Override
  public String toString() {
    return "Resource [source=" + source + ", target=" + target + ", type="
        + type + ", action=" + action + ", owner=" + owner + ", group=" + group
        + ", mode=" + mode + ", recursiveChown=" + recursiveChown
        + ", recursiveChmod=" + recursiveChmod
        + ", changePermissionforParents=" + changePermissionforParents + "]";
  }

  /*
   * Check if parameters are correctly set
   */
  public static void checkResourceParameters(Resource resource,
      FileSystem dfs) throws IllegalArgumentException, IOException {

    ArrayList<String> actionsAvailable = new ArrayList<String>();
    actionsAvailable.add("create");
    actionsAvailable.add("delete");
    ArrayList<String> typesAvailable = new ArrayList<String>();
    typesAvailable.add("file");
    typesAvailable.add("directory");

    if (resource.getTarget() == null)
      throw new IllegalArgumentException(
          "Path to resource in HadoopFs must be filled.");

    if (resource.getAction() == null
        || !actionsAvailable.contains(resource.getAction()))
      throw new IllegalArgumentException("Action is not supported.");

    if (resource.getType() == null
        || !typesAvailable.contains(resource.getType()))
      throw new IllegalArgumentException("Type is not supported.");

    // Check consistency for ("type":"file" == file in hadoop)
    if (dfs.isFile(new Path(resource.getTarget()))
        && !"file".equals(resource.getType()))
      throw new IllegalArgumentException(
          "Cannot create a directory " + resource.getTarget() +
              " because file is present on the given path.");
    // Check consistency for ("type":"directory" == directory in hadoop)
    else if (dfs.isDirectory(new Path(resource.getTarget()))
        && !"directory".equals(resource.getType()))
      throw new IllegalArgumentException(
          "Cannot create a file " + resource.getTarget() +
              " because directory is present on the given path.");
    
    if(resource.getSource() != null) {
      File source = new File(resource.getSource());
      if(source.isFile()
          && !"file".equals(resource.getType()))
        throw new IllegalArgumentException(
            "Cannot create a directory " + resource.getTarget() +
                " because source " + resource.getSource() + "is a file");
      else if(source.isDirectory()
          && !"directory".equals(resource.getType()))
        throw new IllegalArgumentException(
            "Cannot create a file " + resource.getTarget() +
                " because source " + resource.getSource() + "is a directory");      
    }
  }

  /*
   * Create/copy resource - {type}
   */
  public static void createResource(Resource resource,
      FileSystem dfs, Path pathHadoop) throws IOException {

    boolean isCreate = (resource.getSource() == null) ? true : false;

    if (isCreate && resource.getType().equals("directory")) {
      dfs.mkdirs(pathHadoop); // empty dir(s)
    } else if (isCreate && resource.getType().equals("file")) {
      dfs.createNewFile(pathHadoop); // empty file
    } else {
      dfs.copyFromLocalFile(new Path(resource.getSource()), pathHadoop);// copy
    }
  }

  /*
   * Set permissions on resource - {mode}
   */
  public static void setMode(Resource resource,
      FileSystem dfs, Path pathHadoop) throws IOException {

    if (resource.getMode() != null) {
      FsPermission permission = new FsPermission(resource.getMode());
      dfs.setPermission(pathHadoop, permission);

      // Recursive
      
        // Get the list of sub-directories and files
        HashSet<String> resultSet = new HashSet<String>();
        
        if (resource.isRecursiveChmod())
          resource.fillDirectoryList(dfs, resource.getTarget(), resultSet);
        
        if(resource.isChangePermissionOnParents())
          resource.fillInParentDirectories(dfs, resource.getTarget(), resultSet);

        for (String path : resultSet) {
          dfs.setPermission(new Path(path), permission);
        }

    }
  }

  /*
   * Set owner on resource - {owner}
   */
  public static void setOwner(Resource resource, FileSystem dfs,
      Path pathHadoop) throws IOException {

    if (!(resource.getOwner() == null && resource.getGroup() == null)) {
      dfs.setOwner(pathHadoop, resource.getOwner(), resource.getGroup());

      // Get the list of sub-directories and files
      HashSet<String> resultSet = new HashSet<String>();
      if (resource.isRecursiveChown())
        resource.fillDirectoryList(dfs, resource.getTarget(), resultSet);
      if(resource.isChangePermissionOnParents())
        resource.fillInParentDirectories(dfs, resource.getTarget(), resultSet);

      for (String path : resultSet) {
        dfs.setOwner(new Path(path), resource.getOwner(), resource.getGroup());
      }
    }
  }
  
  public void fillInParentDirectories(FileSystem dfs, String path, HashSet<String> resultSet) throws IOException {
    Path filePath = new Path(path);
      
    while(true) {
      filePath = filePath.getParent();
      
      // if(filePath.isRoot()) {
      if(filePath.getParent() == null) {
        break;
      }
      resultSet.add(filePath.toString());
    }
  }

  /*
   * List all files and sub-directories recursively
   */
  public void fillDirectoryList(FileSystem dfs, String path,
      HashSet<String> resultSet) throws IOException {

    FileStatus[] fileStatus = dfs.listStatus(new Path(path));
    if (fileStatus != null) {
      // Go through all resources in directory
      for (FileStatus fs : fileStatus) {
        String pathToResource = path + "/" + fs.getPath().getName();

        resultSet.add(pathToResource);

        if (fs.isDir()) {
          // recursive
          fillDirectoryList(dfs, pathToResource, resultSet);
        }
      }
    }
  }
}
