/**
 * Get the parent of a path
 * @param path path to look at
 * @return the parent path
 * @throws PathNotFoundException if the path was at root.
 */
public static String parentOf(String path) throws PathNotFoundException {
  List<String> elements = split(path);

  int size = elements.size();
  if (size == 0) {
    throw new PathNotFoundException("No parent of " + path);
  }
  if (size == 1) {
    return "/";
  }
  elements.remove(size - 1);
  StringBuilder parent = new StringBuilder(path.length());
  for (String element : elements) {
    parent.append("/");
    parent.append(element);
  }
  return parent.toString();
}
 

/**
 * List children of a directory and retrieve their
 * {@link RegistryPathStatus} values.
 * <p>
 * This is not an atomic operation; A child may be deleted
 * during the iteration through the child entries. If this happens,
 * the <code>PathNotFoundException</code> is caught and that child
 * entry ommitted.
 *
 * @param path path
 * @return a possibly empty map of child entries listed by
 * their short name.
 * @throws PathNotFoundException path is not in the registry.
 * @throws InvalidPathnameException the path is invalid.
 * @throws IOException Any other IO Exception
 */
public static Map<String, RegistryPathStatus> statChildren(
    RegistryOperations registryOperations,
    String path)
    throws PathNotFoundException,
    InvalidPathnameException,
    IOException {
  List<String> childNames = registryOperations.list(path);
  Map<String, RegistryPathStatus> results =
      new HashMap<String, RegistryPathStatus>();
  for (String childName : childNames) {
    String child = join(path, childName);
    try {
      RegistryPathStatus stat = registryOperations.stat(child);
      results.put(childName, stat);
    } catch (PathNotFoundException pnfe) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("stat failed on {}: moved? {}", child, pnfe, pnfe);
      }
      // and continue
    }
  }
  return results;
}
 

/**
 * Stat the file
 * @param path path of operation
 * @return a curator stat entry
 * @throws IOException on a failure
 * @throws PathNotFoundException if the path was not found
 */
public Stat zkStat(String path) throws IOException {
  checkServiceLive();
  String fullpath = createFullPath(path);
  Stat stat;
  try {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Stat {}", fullpath);
    }
    stat = curator.checkExists().forPath(fullpath);
  } catch (Exception e) {
    throw operationFailure(fullpath, "read()", e);
  }
  if (stat == null) {
    throw new PathNotFoundException(path);
  }
  return stat;
}
 

/**
 * Get the ACLs of a path
 * @param path path of operation
 * @return a possibly empty list of ACLs
 * @throws IOException
 */
public List<ACL> zkGetACLS(String path) throws IOException {
  checkServiceLive();
  String fullpath = createFullPath(path);
  List<ACL> acls;
  try {
    if (LOG.isDebugEnabled()) {
      LOG.debug("GetACLS {}", fullpath);
    }
    acls = curator.getACL().forPath(fullpath);
  } catch (Exception e) {
    throw operationFailure(fullpath, "read()", e);
  }
  if (acls == null) {
    throw new PathNotFoundException(path);
  }
  return acls;
}
 

/**
 *  The last arg is expected to be a remote path, if only one argument is
 *  given then the destination will be the remote user's directory 
 *  @param args is the list of arguments
 *  @throws PathIOException if path doesn't exist or matches too many times 
 */
protected void getRemoteDestination(LinkedList<String> args)
throws IOException {
  if (args.size() < 2) {
    dst = new PathData(Path.CUR_DIR, getConf());
  } else {
    String pathString = args.removeLast();
    // if the path is a glob, then it must match one and only one path
    PathData[] items = PathData.expandAsGlob(pathString, getConf());
    switch (items.length) {
      case 0:
        throw new PathNotFoundException(pathString);
      case 1:
        dst = items[0];
        break;
      default:
        throw new PathIOException(pathString, "Too many matches");
    }
  }
}
 

@Override
protected void processArguments(LinkedList<PathData> args)
throws IOException {
  // if more than one arg, the destination must be a directory
  // if one arg, the dst must not exist or must be a directory
  if (args.size() > 1) {
    if (!dst.exists) {
      throw new PathNotFoundException(dst.toString());
    }
    if (!dst.stat.isDirectory()) {
      throw new PathIsNotDirectoryException(dst.toString());
    }
  } else if (dst.exists) {
    if (!dst.stat.isDirectory() && !overwrite) {
      throw new PathExistsException(dst.toString());
    }
  } else if (!dst.parentExists()) {
    throw new PathNotFoundException(dst.toString());
  }
  super.processArguments(args);
}
 

/**
 * Get the parent of a path
 * @param path path to look at
 * @return the parent path
 * @throws PathNotFoundException if the path was at root.
 */
public static String parentOf(String path) throws PathNotFoundException {
  List<String> elements = split(path);

  int size = elements.size();
  if (size == 0) {
    throw new PathNotFoundException("No parent of " + path);
  }
  if (size == 1) {
    return "/";
  }
  elements.remove(size - 1);
  StringBuilder parent = new StringBuilder(path.length());
  for (String element : elements) {
    parent.append("/");
    parent.append(element);
  }
  return parent.toString();
}
 

/**
 * List children of a directory and retrieve their
 * {@link RegistryPathStatus} values.
 * <p>
 * This is not an atomic operation; A child may be deleted
 * during the iteration through the child entries. If this happens,
 * the <code>PathNotFoundException</code> is caught and that child
 * entry ommitted.
 *
 * @param path path
 * @return a possibly empty map of child entries listed by
 * their short name.
 * @throws PathNotFoundException path is not in the registry.
 * @throws InvalidPathnameException the path is invalid.
 * @throws IOException Any other IO Exception
 */
public static Map<String, RegistryPathStatus> statChildren(
    RegistryOperations registryOperations,
    String path)
    throws PathNotFoundException,
    InvalidPathnameException,
    IOException {
  List<String> childNames = registryOperations.list(path);
  Map<String, RegistryPathStatus> results =
      new HashMap<String, RegistryPathStatus>();
  for (String childName : childNames) {
    String child = join(path, childName);
    try {
      RegistryPathStatus stat = registryOperations.stat(child);
      results.put(childName, stat);
    } catch (PathNotFoundException pnfe) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("stat failed on {}: moved? {}", child, pnfe, pnfe);
      }
      // and continue
    }
  }
  return results;
}
 

/**
 * Stat the file
 * @param path path of operation
 * @return a curator stat entry
 * @throws IOException on a failure
 * @throws PathNotFoundException if the path was not found
 */
public Stat zkStat(String path) throws IOException {
  checkServiceLive();
  String fullpath = createFullPath(path);
  Stat stat;
  try {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Stat {}", fullpath);
    }
    stat = curator.checkExists().forPath(fullpath);
  } catch (Exception e) {
    throw operationFailure(fullpath, "read()", e);
  }
  if (stat == null) {
    throw new PathNotFoundException(path);
  }
  return stat;
}
 

/**
 * Get the ACLs of a path
 * @param path path of operation
 * @return a possibly empty list of ACLs
 * @throws IOException
 */
public List<ACL> zkGetACLS(String path) throws IOException {
  checkServiceLive();
  String fullpath = createFullPath(path);
  List<ACL> acls;
  try {
    if (LOG.isDebugEnabled()) {
      LOG.debug("GetACLS {}", fullpath);
    }
    acls = curator.getACL().forPath(fullpath);
  } catch (Exception e) {
    throw operationFailure(fullpath, "read()", e);
  }
  if (acls == null) {
    throw new PathNotFoundException(path);
  }
  return acls;
}
 

/**
 *  The last arg is expected to be a remote path, if only one argument is
 *  given then the destination will be the remote user's directory 
 *  @param args is the list of arguments
 *  @throws PathIOException if path doesn't exist or matches too many times 
 */
protected void getRemoteDestination(LinkedList<String> args)
throws IOException {
  if (args.size() < 2) {
    dst = new PathData(Path.CUR_DIR, getConf());
  } else {
    String pathString = args.removeLast();
    // if the path is a glob, then it must match one and only one path
    PathData[] items = PathData.expandAsGlob(pathString, getConf());
    switch (items.length) {
      case 0:
        throw new PathNotFoundException(pathString);
      case 1:
        dst = items[0];
        break;
      default:
        throw new PathIOException(pathString, "Too many matches");
    }
  }
}
 

@Override
protected void processArguments(LinkedList<PathData> args)
throws IOException {
  // if more than one arg, the destination must be a directory
  // if one arg, the dst must not exist or must be a directory
  if (args.size() > 1) {
    if (!dst.exists) {
      throw new PathNotFoundException(dst.toString());
    }
    if (!dst.stat.isDirectory()) {
      throw new PathIsNotDirectoryException(dst.toString());
    }
  } else if (dst.exists) {
    if (!dst.stat.isDirectory() && !overwrite) {
      throw new PathExistsException(dst.toString());
    }
  } else if (!dst.parentExists()) {
    throw new PathNotFoundException(dst.toString());
  }
  super.processArguments(args);
}
 

/**
 * Translates HDFS specific Exceptions to Pravega-equivalent Exceptions.
 *
 * @param segmentName Name of the stream segment on which the exception occurs.
 * @param e           The exception to be translated.
 * @return  The exception to be thrown.
 */
static <T> StreamSegmentException convertException(String segmentName, Throwable e) {
    if (e instanceof RemoteException) {
        e = ((RemoteException) e).unwrapRemoteException();
    }

    if (e instanceof PathNotFoundException || e instanceof FileNotFoundException) {
        return new StreamSegmentNotExistsException(segmentName, e);
    } else if (e instanceof FileAlreadyExistsException || e instanceof AlreadyBeingCreatedException) {
        return new StreamSegmentExistsException(segmentName, e);
    } else if (e instanceof AclException) {
        return new StreamSegmentSealedException(segmentName, e);
    } else {
        throw Exceptions.sneakyThrow(e);
    }
}
 

/**
 * Given an exception and a possibly empty argument list, generate
 * a diagnostics string for use in error messages
 * @param operation the operation that failed
 * @param e exception
 * @param argsList arguments list
 * @return a string intended for the user
 */
String analyzeException(String operation,
    Exception e,
    List<String> argsList) {

  String pathArg = !argsList.isEmpty() ? argsList.get(1) : "(none)";
  if (LOG.isDebugEnabled()) {
    LOG.debug("Operation {} on path {} failed with exception {}",
        operation, pathArg, e, e);
  }
  if (e instanceof InvalidPathnameException) {
    return "InvalidPath :" + pathArg + ": " + e;
  }
  if (e instanceof PathNotFoundException) {
    return "Path not found: " + pathArg;
  }
  if (e instanceof NoRecordException) {
    return "No service record at path " + pathArg;
  }
  if (e instanceof AuthenticationFailedException) {
    return "Failed to authenticate to registry : " + e;
  }
  if (e instanceof NoPathPermissionsException) {
    return "No Permission to path: " + pathArg + ": " + e;
  }
  if (e instanceof AccessControlException) {
    return "No Permission to path: " + pathArg + ": " + e;
  }
  if (e instanceof InvalidRecordException) {
    return "Unable to read record at: " + pathArg + ": " + e;
  }
  if (e instanceof IOException) {
    return "IO Exception when accessing path :" + pathArg + ": " + e;
  }
  // something else went very wrong here
  return "Exception " + e;

}
 

@Test
public void testMkdirNoParent() throws Throwable {
  String path = ENTRY_PATH + "/missing";
  try {
    operations.mknode(path, false);
    RegistryPathStatus stat = operations.stat(path);
    fail("Got a status " + stat);
  } catch (PathNotFoundException expected) {
    // expected
  }
}
 

@Test
public void testPutNoParent() throws Throwable {
  ServiceRecord record = new ServiceRecord();
  record.set(YarnRegistryAttributes.YARN_ID, "testPutNoParent");
  String path = "/path/without/parent";
  try {
    operations.bind(path, record, 0);
    // didn't get a failure
    // trouble
    RegistryPathStatus stat = operations.stat(path);
    fail("Got a status " + stat);
  } catch (PathNotFoundException expected) {
    // expected
  }
}
 

@Test(expected = PathNotFoundException.class)
public void testPutNoParent2() throws Throwable {
  ServiceRecord record = new ServiceRecord();
  record.set(YarnRegistryAttributes.YARN_ID, "testPutNoParent");
  String path = "/path/without/parent";
  operations.bind(path, record, 0);
}
 

/**
 * assert that a path does not exist
 * @param path path in the registry
 * @throws IOException
 */
public void assertPathNotFound(String path) throws IOException {
  try {
    operations.stat(path);
    fail("Path unexpectedly found: " + path);
  } catch (PathNotFoundException e) {

  }
}
 

@Override
protected List<PathData> expandArgument(String arg) throws IOException {
  try {
    return super.expandArgument(arg);
  } catch (PathNotFoundException e) {
    if (!ignoreFNF) {
      throw e;
    }
    // prevent -f on a non-existent glob from failing
    return new LinkedList<PathData>();
  }
}
 

@Override
protected void processNonexistentPath(PathData item) throws IOException {
  if (!item.parentExists()) {
    throw new PathNotFoundException(item.toString());
  }
  touchz(item);
}
 

/**
 * Expand the given argument into a list of {@link PathData} objects.
 * The default behavior is to expand globs.  Commands may override to
 * perform other expansions on an argument.
 * @param arg string pattern to expand
 * @return list of {@link PathData} objects
 * @throws IOException if anything goes wrong...
 */
protected List<PathData> expandArgument(String arg) throws IOException {
  PathData[] items = PathData.expandAsGlob(arg, getConf());
  if (items.length == 0) {
    // it's a glob that failed to match
    throw new PathNotFoundException(arg);
  }
  return Arrays.asList(items);
}
 

/**
 * Get the FileStatus info
 * @param ignoreFNF if true, stat will be null if the path doesn't exist
 * @return FileStatus for the given path
 * @throws IOException if anything goes wrong
 */
private static
FileStatus lookupStat(FileSystem fs, String pathString, boolean ignoreFNF)
throws IOException {
  FileStatus status = null;
  try {
    status = fs.getFileStatus(new Path(pathString));
  } catch (FileNotFoundException e) {
    if (!ignoreFNF) throw new PathNotFoundException(pathString);
  }
  // TODO: should consider wrapping other exceptions into Path*Exceptions
  return status;
}
 

/**
 * Ensure that the file exists and if it is or is not a directory
 * @param typeRequirement Set it to the desired requirement.
 * @throws PathIOException if file doesn't exist or the type does not match
 * what was specified in typeRequirement.
 */
private void checkIfExists(FileTypeRequirement typeRequirement) 
throws PathIOException {
  if (!exists) {
    throw new PathNotFoundException(toString());      
  }

  if ((typeRequirement == FileTypeRequirement.SHOULD_BE_DIRECTORY)
     && !stat.isDirectory()) {
    throw new PathIsNotDirectoryException(toString());
  } else if ((typeRequirement == FileTypeRequirement.SHOULD_NOT_BE_DIRECTORY)
            && stat.isDirectory()) {
    throw new PathIsDirectoryException(toString());
  }
}
 

@Override
protected void processNonexistentPath(PathData item) throws IOException {
  // check if parent exists. this is complicated because getParent(a/b/c/) returns a/b/c, but
  // we want a/b
  if (!item.fs.exists(new Path(item.path.toString()).getParent()) && !createParents) {
    throw new PathNotFoundException(item.toString());
  }
  if (!item.fs.mkdirs(item.path)) {
    throw new PathIOException(item.toString());
  }
}
 

/**
 * Given an exception and a possibly empty argument list, generate
 * a diagnostics string for use in error messages
 * @param operation the operation that failed
 * @param e exception
 * @param argsList arguments list
 * @return a string intended for the user
 */
String analyzeException(String operation,
    Exception e,
    List<String> argsList) {

  String pathArg = !argsList.isEmpty() ? argsList.get(1) : "(none)";
  if (LOG.isDebugEnabled()) {
    LOG.debug("Operation {} on path {} failed with exception {}",
        operation, pathArg, e, e);
  }
  if (e instanceof InvalidPathnameException) {
    return "InvalidPath :" + pathArg + ": " + e;
  }
  if (e instanceof PathNotFoundException) {
    return "Path not found: " + pathArg;
  }
  if (e instanceof NoRecordException) {
    return "No service record at path " + pathArg;
  }
  if (e instanceof AuthenticationFailedException) {
    return "Failed to authenticate to registry : " + e;
  }
  if (e instanceof NoPathPermissionsException) {
    return "No Permission to path: " + pathArg + ": " + e;
  }
  if (e instanceof AccessControlException) {
    return "No Permission to path: " + pathArg + ": " + e;
  }
  if (e instanceof InvalidRecordException) {
    return "Unable to read record at: " + pathArg + ": " + e;
  }
  if (e instanceof IOException) {
    return "IO Exception when accessing path :" + pathArg + ": " + e;
  }
  // something else went very wrong here
  return "Exception " + e;

}
 

@Test
public void testMkdirNoParent() throws Throwable {
  String path = ENTRY_PATH + "/missing";
  try {
    operations.mknode(path, false);
    RegistryPathStatus stat = operations.stat(path);
    fail("Got a status " + stat);
  } catch (PathNotFoundException expected) {
    // expected
  }
}
 

@Test
public void testPutNoParent() throws Throwable {
  ServiceRecord record = new ServiceRecord();
  record.set(YarnRegistryAttributes.YARN_ID, "testPutNoParent");
  String path = "/path/without/parent";
  try {
    operations.bind(path, record, 0);
    // didn't get a failure
    // trouble
    RegistryPathStatus stat = operations.stat(path);
    fail("Got a status " + stat);
  } catch (PathNotFoundException expected) {
    // expected
  }
}
 

@Test(expected = PathNotFoundException.class)
public void testPutNoParent2() throws Throwable {
  ServiceRecord record = new ServiceRecord();
  record.set(YarnRegistryAttributes.YARN_ID, "testPutNoParent");
  String path = "/path/without/parent";
  operations.bind(path, record, 0);
}
 

/**
 * assert that a path does not exist
 * @param path path in the registry
 * @throws IOException
 */
public void assertPathNotFound(String path) throws IOException {
  try {
    operations.stat(path);
    fail("Path unexpectedly found: " + path);
  } catch (PathNotFoundException e) {

  }
}
 

@Override
protected List<PathData> expandArgument(String arg) throws IOException {
  try {
    return super.expandArgument(arg);
  } catch (PathNotFoundException e) {
    if (!ignoreFNF) {
      throw e;
    }
    // prevent -f on a non-existent glob from failing
    return new LinkedList<PathData>();
  }
}