public FsDirectory(FileSystem fs,
    Path directory,
    boolean create,
    Configuration conf) throws IOException 
{
    this.fs = fs;
    this.directory = directory;
    this.ioFileBufferSize = conf.getInt("io.file.buffer.size", 4096);
    if (create) {
      create();
    }
    if (!fs.getFileStatus(directory).isDir())
      throw new IOException(directory + " not a directory");
}

 public synchronized  void close() throws IOException 
{
    fs.close();
}

 public IndexOutput createOutput(String name) throws IOException 
{
    Path file = new Path(directory, name);
    if (fs.exists(file) && !fs.delete(file, false))      // delete existing, if any
      throw new IOException("Cannot overwrite: " + file);
    return new DfsIndexOutput(file, this.ioFileBufferSize);
}

 public  void deleteFile(String name) throws IOException 
{
    if (!fs.delete(new Path(directory, name), false))
      throw new IOException("Cannot delete " + name);
}

 public boolean fileExists(String name) throws IOException 
{
    return fs.exists(new Path(directory, name));
}

 public long fileLength(String name) throws IOException 
{
    return fs.getFileStatus(new Path(directory, name)).getLen();
}

 public long fileModified(String name) 
{
    throw new UnsupportedOperationException();
}

 public String[] list() throws IOException 
{
    FileStatus[] fstats = fs.listStatus(directory, HadoopFSUtil.getPassAllFilter());
    Path[] files = HadoopFSUtil.getPaths(fstats);
    if (files == null) return null;
    String[] result = new String[files.length];
    for (int i = 0; i <  files.length; i++) {
      result[i] = files[i].getName();
    }
    return result;
}

 public Lock makeLock(String name) 
{
    return new Lock() {
      public boolean obtain() {
        return true;
      }
      public void release() {
      }
      public boolean isLocked() {
        throw new UnsupportedOperationException();
      }
      public String toString() {
        return "Lock@" + new Path(directory, name);
      }
    };
}

 public IndexInput openInput(String name) throws IOException 
{
    return new DfsIndexInput(new Path(directory, name), this.ioFileBufferSize);
}

 public  void renameFile(String from,
    String to) throws IOException 
{
    // DFS is currently broken when target already exists,
    // so we explicitly delete the target first.
    Path target = new Path(directory, to);
    if (fs.exists(target)) {
      fs.delete(target, false);
    }
    fs.rename(new Path(directory, from), target);
}

 public String toString() 
{
    return this.getClass().getName() + "@" + directory;
}

 public  void touchFile(String name) 
{
    throw new UnsupportedOperationException();
}