text: /**
   * 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.
   */
  
  #include "hdfs.h"
  #include "hdfsJniHelper.h"
  
  
  /* Some frequently used Java paths */
  #define HADOOP_CONF     "org/apache/hadoop/conf/Configuration"
  #define HADOOP_PATH     "org/apache/hadoop/fs/Path"
  #define HADOOP_LOCALFS  "org/apache/hadoop/fs/LocalFileSystem"
  #define HADOOP_FS       "org/apache/hadoop/fs/FileSystem"
  #define HADOOP_BLK_LOC  "org/apache/hadoop/fs/BlockLocation"
  #define HADOOP_DFS      "org/apache/hadoop/dfs/DistributedFileSystem"
  #define HADOOP_ISTRM    "org/apache/hadoop/fs/FSDataInputStream"
  #define HADOOP_OSTRM    "org/apache/hadoop/fs/FSDataOutputStream"
  #define HADOOP_STAT     "org/apache/hadoop/fs/FileStatus"
  #define JAVA_NET_ISA    "java/net/InetSocketAddress"
  #define JAVA_NET_URI    "java/net/URI"
  
  
  
  /* Macros for constructing method signatures */
  #define JPARAM(X)           "L" X ";"
  #define JARRPARAM(X)        "[L" X ";"
  #define JMETHOD1(X, R)      "(" X ")" R
  #define JMETHOD2(X, Y, R)   "(" X Y ")" R
  
  
  /**
   * hdfsJniEnv: A wrapper struct to be used as 'value'
   * while saving thread -> JNIEnv* mappings
   */
  typedef struct
  {
      JNIEnv* env;
  } hdfsJniEnv;
  
  
  
  /**
   * Helper function to destroy a local reference of java.lang.Object
   * @param env: The JNIEnv pointer. 
   * @param jFile: The local reference of java.lang.Object object
   * @return None.
   */
  static void destroyLocalReference(JNIEnv *env, jobject jObject)
  {
    if (jObject)
      (*env)->DeleteLocalRef(env, jObject);
  }
  
  
  /**
   * Helper function to create a org.apache.hadoop.fs.Path object.
   * @param env: The JNIEnv pointer. 
   * @param path: The file-path for which to construct org.apache.hadoop.fs.Path
   * object.
   * @return Returns a jobject on success and NULL on error.
   */
  static jobject constructNewObjectOfPath(JNIEnv *env, const char *path)
  {
      //Construct a java.lang.String object
      jstring jPathString = (*env)->NewStringUTF(env, path); 
  
      //Construct the org.apache.hadoop.fs.Path object
      jobject jPath =
          constructNewObjectOfClass(env, "org/apache/hadoop/fs/Path",
                                    "(Ljava/lang/String;)V", jPathString);
      if (jPath == NULL) {
          fprintf(stderr, "Can't construct instance of class "
                  "org.apache.hadoop.fs.Path for %s\n", path);
          errno = EINTERNAL;
          return NULL;
      }
  
      // Destroy the local reference to the java.lang.String object
      destroyLocalReference(env, jPathString);
  
      return jPath;
  }
  
  
  
  
  
  
  hdfsFS hdfsConnect(const char* host, tPort port)
  {
      // JAVA EQUIVALENT:
      //  FileSystem fs = FileSystem.get(new Configuration());
      //  return fs;
  
      JNIEnv *env = 0;
      jobject jConfiguration = NULL;
      jobject jFS = NULL;
      jobject jURI = NULL;
      jstring jURIString = NULL;
      jvalue  jVal;
      char    *cURI = 0;
      jobject gFsRef = NULL;
  
  
      //Get the JNIEnv* corresponding to current thread
      env = getJNIEnv();
  
      //Create the org.apache.hadoop.conf.Configuration object
      jConfiguration =
          constructNewObjectOfClass(env, HADOOP_CONF, "()V");
  
      if (jConfiguration == NULL) {
          fprintf(stderr, "Can't construct instance of class "
                  "org.apache.hadoop.conf.Configuration\n");
          errno = EINTERNAL;
          return NULL;
      }
   
      //Check what type of FileSystem the caller wants...
      if (host == NULL) {
          // fs = FileSytem::getLocal(conf);
          if (invokeMethod(env, &jVal, STATIC, NULL, HADOOP_FS, "getLocal",
                           JMETHOD1(JPARAM(HADOOP_CONF),
                                    JPARAM(HADOOP_LOCALFS)),
                           jConfiguration) != 0) {
            fprintf(stderr, "Call to org.apache.hadoop.fs."
                    "FileSystem::getLocal failed!\n");
            errno = EINTERNAL;
            goto done;
          }
          jFS = jVal.l;
      }
      else if (!strcmp(host, "default") && port == 0) {
          //fs = FileSystem::get(conf); 
          if (invokeMethod(env, &jVal, STATIC, NULL,
                           HADOOP_FS, "get",
                           JMETHOD1(JPARAM(HADOOP_CONF),
                                    JPARAM(HADOOP_FS)),
                           jConfiguration) != 0) {
              fprintf(stderr, "Call to org.apache.hadoop.fs."
                      "FileSystem::get failed!\n");
              errno = EINTERNAL;
              goto done;
          }
          jFS = jVal.l;
      }
      else {
          // fs = FileSystem::get(URI, conf);
          cURI = malloc(strlen(host)+16);
          sprintf(cURI, "hdfs://%s:%d", host, (int)(port));
  
          jURIString = (*env)->NewStringUTF(env, cURI);
          if (invokeMethod(env, &jVal, STATIC, NULL, JAVA_NET_URI,
                           "create", "(Ljava/lang/String;)Ljava/net/URI;",
                           jURIString) != 0) {
            fprintf(stderr, "Call to java.net.URI::create failed!\n");
            errno = EINTERNAL;
            goto done;
          }
          jURI = jVal.l;
  
          if (invokeMethod(env, &jVal, STATIC, NULL, HADOOP_FS, "get",
                           JMETHOD2(JPARAM(JAVA_NET_URI),
                                    JPARAM(HADOOP_CONF), JPARAM(HADOOP_FS)),
                           jURI, jConfiguration) != 0) {
            fprintf(stderr, "Call to org.apache.hadoop.fs."
                    "FileSystem::get(URI, Configuration) failed!\n");
            errno = EINTERNAL;
            goto done;
          }
  
          jFS = jVal.l;
      }
  
    done:
      
      // Release unnecessary local references
      destroyLocalReference(env, jConfiguration);
      destroyLocalReference(env, jURIString);
      destroyLocalReference(env, jURI);
  
      if (cURI) free(cURI);
  
      /* Create a global reference for this fs */
      if (jFS) {
          gFsRef = (*env)->NewGlobalRef(env, jFS);
          destroyLocalReference(env, jFS);
      }
  
      return gFsRef;
  }
  
  
  
  int hdfsDisconnect(hdfsFS fs)
  {
      // JAVA EQUIVALENT:
      //  fs.close()
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jFS = (jobject)fs;
  
      //Sanity check
      if (fs == NULL) {
          errno = EBADF;
          return -1;
      }
  
      if (invokeMethod(env, NULL, INSTANCE, jFS, HADOOP_FS,
                       "close", "()V") != 0) {
          fprintf(stderr, "Call to FileSystem::close failed!\n"); 
          errno = EINTERNAL;
          return -1;
      }
  
      //Release unnecessary references
      (*env)->DeleteGlobalRef(env, fs);
  
      return 0;
  }
  
  
  
  hdfsFile hdfsOpenFile(hdfsFS fs, const char* path, int flags, 
                        int bufferSize, short replication, tSize blockSize)
  {
      /*
        JAVA EQUIVALENT:
         File f = new File(path);
         FSData{Input|Output}Stream f{is|os} = fs.create(f);
         return f{is|os};
      */
  
      /* Get the JNIEnv* corresponding to current thread */
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      /* The hadoop java api/signature */
      const char* method = (flags == O_RDONLY) ? "open" : "create";
      const char* signature = (flags == O_RDONLY) ?
          JMETHOD2(JPARAM(HADOOP_PATH), "I", JPARAM(HADOOP_ISTRM)) :
          JMETHOD2(JPARAM(HADOOP_PATH), "ZISJ", JPARAM(HADOOP_OSTRM));
  
      /* Return value */
      hdfsFile file = NULL;
  
      /* Create an object of org.apache.hadoop.fs.Path */
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return NULL; 
      }
  
      /* Get the Configuration object from the FileSystem object */
      jvalue  jVal;
      jobject jConfiguration = NULL;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "getConf", JMETHOD1("", JPARAM(HADOOP_CONF))) != 0) {
          fprintf(stderr, "Failed to get configuration object from "
                  "filesystem\n");
          errno = EINTERNAL;
          destroyLocalReference(env, jPath);
          return NULL;
      }
      jConfiguration = jVal.l;
  
      jint jBufferSize = bufferSize;
      jshort jReplication = replication;
      jlong jBlockSize = blockSize;
      jstring jStrBufferSize = (*env)->NewStringUTF(env, "io.file.buffer.size"); 
      jstring jStrReplication = (*env)->NewStringUTF(env, "dfs.replication");
      jstring jStrBlockSize = (*env)->NewStringUTF(env, "dfs.block.size");
  
  
      //bufferSize
      if (!bufferSize) {
          if (invokeMethod(env, &jVal, INSTANCE, jConfiguration, 
                           HADOOP_CONF, "getInt", "(Ljava/lang/String;I)I",
                           jStrBufferSize, 4096) != 0) {
              fprintf(stderr, "Call to org.apache.hadoop.conf."
                      "Configuration::getInt failed!\n");
              errno = EINTERNAL;
              goto done;
          }
          jBufferSize = jVal.i;
      }
  
      if (flags == O_WRONLY) {
          //replication
  
          if (!replication) {
              if (invokeMethod(env, &jVal, INSTANCE, jConfiguration, 
                               HADOOP_CONF, "getInt", "(Ljava/lang/String;I)I",
                               jStrReplication, 1) != 0) {
                  fprintf(stderr, "Call to org.apache.hadoop.conf."
                          "Configuration::getInt failed!\n");
                  errno = EINTERNAL;
                  goto done;
              }
              jReplication = jVal.i;
          }
          
          //blockSize
          if (!blockSize) {
              if (invokeMethod(env, &jVal, INSTANCE, jConfiguration, 
                               HADOOP_CONF, "getLong", "(Ljava/lang/String;J)J",
                               jStrBlockSize, 67108864)) {
                  fprintf(stderr, "Call to org.apache.hadoop.fs."
                          "FileSystem::%s(%s) failed!\n", method, signature);
                  errno = EINTERNAL;
                  goto done;
              }
              jBlockSize = jVal.j;
          }
      }
   
      /* Create and return either the FSDataInputStream or
         FSDataOutputStream references jobject jStream */
  
      if (flags == O_RDONLY) {
          if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                           method, signature, jPath, jBufferSize)) {
              fprintf(stderr, "Call to org.apache.hadoop.fs."
                      "FileSystem::%s(%s) failed!\n", method, signature);
              errno = EINTERNAL;
              goto done;
          }
      }
      else {
          jboolean jOverWrite = 1;
          if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                           method, signature, jPath, jOverWrite,
                           jBufferSize, jReplication, jBlockSize)) {
              fprintf(stderr, "Call to org.apache.hadoop.fs."
                      "FileSystem::%s(%s) failed!\n", method, signature);
              errno = EINTERNAL;
              goto done;
          }
      }
    
      file = malloc(sizeof(struct hdfsFile_internal));
      if (!file) {
          errno = ENOMEM;
          return NULL;
      }
      file->file = (*env)->NewGlobalRef(env, jVal.l);
      file->type = ((flags == O_RDONLY) ? INPUT : OUTPUT);
  
      destroyLocalReference(env, jVal.l);
  
      done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jStrBufferSize);
      destroyLocalReference(env, jStrReplication);
      destroyLocalReference(env, jStrBlockSize);
      destroyLocalReference(env, jConfiguration); 
      destroyLocalReference(env, jPath); 
  
      return file;
  }
  
  
  
  int hdfsCloseFile(hdfsFS fs, hdfsFile file)
  {
      // JAVA EQUIVALENT:
      //  file.close 
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jStream = (jobject)(file ? file->file : NULL);
  
      //Sanity check
      if (!file || file->type == UNINITIALIZED) {
          errno = EBADF;
          return -1;
      }
  
      //The interface whose 'close' method to be called
      const char* interface = (file->type == INPUT) ? 
          HADOOP_ISTRM : HADOOP_OSTRM;
    
      if (invokeMethod(env, NULL, INSTANCE, jStream, interface,
                       "close", "()V") != 0) {
          fprintf(stderr, "Call to %s::close failed!\n", interface); 
          errno = EINTERNAL;
          return -1;
      }
  
      //De-allocate memory
      free(file);
      (*env)->DeleteGlobalRef(env, jStream);
  
      return 0;
  }
  
  
  
  int hdfsExists(hdfsFS fs, const char *path)
  {
      JNIEnv *env = getJNIEnv();
      jobject jPath = constructNewObjectOfPath(env, path);
      jvalue  jVal;
      jobject jFS = (jobject)fs;
  
      if (jPath == NULL) {
          return -1;
      }
  
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "exists", JMETHOD1(JPARAM(HADOOP_PATH), "Z"),
                       jPath) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::exists failed!\n"); 
          errno = EINTERNAL;
          return -1;
      }
  
      return jVal.z ? 0 : -1;
  }
  
  
  
  tSize hdfsRead(hdfsFS fs, hdfsFile f, void* buffer, tSize length)
  {
      // JAVA EQUIVALENT:
      //  byte [] bR = new byte[length];
      //  fis.read(bR);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jInputStream = (jobject)(f ? f->file : NULL);
  
      jbyteArray jbRarray;
      jint noReadBytes = 0;
      jvalue jVal;
  
      //Sanity check
      if (!f || f->type == UNINITIALIZED) {
          errno = EBADF;
          return -1;
      }
  
      //Error checking... make sure that this file is 'readable'
      if (f->type != INPUT) {
          fprintf(stderr, "Cannot read from a non-InputStream object!\n");
          errno = EINVAL;
          return -1;
      }
  
      //Read the requisite bytes
      jbRarray = (*env)->NewByteArray(env, length);
      if (invokeMethod(env, &jVal, INSTANCE, jInputStream, HADOOP_ISTRM,
                       "read", "([B)I", jbRarray) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FSDataInputStream::read failed!\n");
          errno = EINTERNAL;
          noReadBytes = -1;
      }
      else {
          noReadBytes = jVal.i;
          if (noReadBytes > 0) {
              (*env)->GetByteArrayRegion(env, jbRarray, 0, noReadBytes, buffer);
          }  else {
              //This is a valid case: there aren't any bytes left to read!
              noReadBytes = 0;
          }
          errno = 0;
      }
  
      destroyLocalReference(env, jbRarray);
  
      return noReadBytes;
  }
  
  
    
  tSize hdfsPread(hdfsFS fs, hdfsFile f, tOffset position,
                  void* buffer, tSize length)
  {
      // JAVA EQUIVALENT:
      //  byte [] bR = new byte[length];
      //  fis.read(pos, bR, 0, length);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jInputStream = (jobject)(f ? f->file : NULL);
  
      jbyteArray jbRarray;
      jint noReadBytes = 0;
      jvalue jVal;
  
      //Sanity check
      if (!f || f->type == UNINITIALIZED) {
          errno = EBADF;
          return -1;
      }
  
      //Error checking... make sure that this file is 'readable'
      if (f->type != INPUT) {
          fprintf(stderr, "Cannot read from a non-InputStream object!\n");
          errno = EINVAL;
          return -1;
      }
  
      //Read the requisite bytes
      jbRarray = (*env)->NewByteArray(env, length);
      if (invokeMethod(env, &jVal, INSTANCE, jInputStream, HADOOP_ISTRM,
                       "read", "(J[BII)I", position, jbRarray, 0, length) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FSDataInputStream::read failed!\n");
          errno = EINTERNAL;
          noReadBytes = -1;
      }
      else {
          noReadBytes = jVal.i;
          if (noReadBytes > 0) {
              (*env)->GetByteArrayRegion(env, jbRarray, 0, noReadBytes, buffer);
          }  else {
              //This is a valid case: there aren't any bytes left to read!
              noReadBytes = 0;
          }
          errno = 0;
      }
      destroyLocalReference(env, jbRarray);
  
      return noReadBytes;
  }
  
  
  
  tSize hdfsWrite(hdfsFS fs, hdfsFile f, const void* buffer, tSize length)
  {
      // JAVA EQUIVALENT
      // byte b[] = str.getBytes();
      // fso.write(b);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jOutputStream = (jobject)(f ? f->file : 0);
      jbyteArray jbWarray;
  
      //Sanity check
      if (!f || f->type == UNINITIALIZED) {
          errno = EBADF;
          return -1;
      }
      
      if (length < 0) {
      	errno = EINVAL;
      	return -1;
      }
  
      //Error checking... make sure that this file is 'writable'
      if (f->type != OUTPUT) {
          fprintf(stderr, "Cannot write into a non-OutputStream object!\n");
          errno = EINVAL;
          return -1;
      }
  
      // 'length' equals 'zero' is a valid use-case according to Posix!
      if (length != 0) {
          //Write the requisite bytes into the file
          jbWarray = (*env)->NewByteArray(env, length);
          (*env)->SetByteArrayRegion(env, jbWarray, 0, length, buffer);
          if (invokeMethod(env, NULL, INSTANCE, jOutputStream,
                           HADOOP_OSTRM, "write",
                           "([B)V", jbWarray) != 0) {
              fprintf(stderr, "Call to org.apache.hadoop.fs."
                      "FSDataOutputStream::write failed!\n");
              errno = EINTERNAL;
              length = -1;
          }
          destroyLocalReference(env, jbWarray);
      }
  
      //Return no. of bytes succesfully written (libc way)
      //i.e. 'length' itself! ;-)
      return length;
  }
  
  
  
  int hdfsSeek(hdfsFS fs, hdfsFile f, tOffset desiredPos) 
  {
      // JAVA EQUIVALENT
      //  fis.seek(pos);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jInputStream = (jobject)(f ? f->file : 0);
  
      //Sanity check
      if (!f || f->type != INPUT) {
          errno = EBADF;
          return -1;
      }
  
      if (invokeMethod(env, NULL, INSTANCE, jInputStream, HADOOP_ISTRM,
                       "seek", "(J)V", desiredPos) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FSDataInputStream::seek failed!\n");
          errno = EINTERNAL;
          return -1;
      }
  
      return 0;
  }
  
  
  
  tOffset hdfsTell(hdfsFS fs, hdfsFile f)
  {
      // JAVA EQUIVALENT
      //  pos = f.getPos();
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jStream = (jobject)(f ? f->file : 0);
  
      //Sanity check
      if (!f || f->type == UNINITIALIZED) {
          errno = EBADF;
          return -1;
      }
  
      const char* interface = (f->type == INPUT) ?
          HADOOP_ISTRM : HADOOP_OSTRM;
  
      jlong currentPos  = -1;
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jStream,
                       interface, "getPos", "()J") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FSDataInputStream::getPos failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      currentPos = jVal.j;
  
      return (tOffset)currentPos;
  }
  
  
  
  int hdfsFlush(hdfsFS fs, hdfsFile f) 
  {
      // JAVA EQUIVALENT
      //  fos.flush();
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jOutputStream = (jobject)(f ? f->file : 0);
  
      //Sanity check
      if (!f || f->type != OUTPUT) {
          errno = EBADF;
          return -1;
      }
  
      if (invokeMethod(env, NULL, INSTANCE, jOutputStream, 
                       HADOOP_OSTRM, "flush", "()V") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FSDataInputStream::flush failed!\n");
          errno = EINTERNAL;
          return -1;
      }
  
      return 0;
  }
  
  
  
  int hdfsAvailable(hdfsFS fs, hdfsFile f)
  {
      // JAVA EQUIVALENT
      //  fis.available();
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jInputStream = (jobject)(f ? f->file : 0);
  
      //Sanity check
      if (!f || f->type != INPUT) {
          errno = EBADF;
          return -1;
      }
  
      jint available = -1;
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jInputStream, 
                       HADOOP_ISTRM, "available", "()I") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FSDataInputStream::available failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      available = jVal.i;
  
      return available;
  }
  
  
  
  int hdfsCopy(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst)
  {
      //JAVA EQUIVALENT
      //  FileUtil::copy(srcFS, srcPath, dstFS, dstPath,
      //                 deleteSource = false, conf)
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jSrcFS = (jobject)srcFS;
      jobject jDstFS = (jobject)dstFS;
      jobject jSrcPath = NULL;
      jobject jDstPath = NULL;
  
      jSrcPath = constructNewObjectOfPath(env, src);
      if (jSrcPath == NULL) {
          return -1;
      }
  
      jDstPath = constructNewObjectOfPath(env, dst);
      if (jDstPath == NULL) {
          destroyLocalReference(env, jSrcPath);
          return -1;
      }
  
      int retval = 0;
  
      //Create the org.apache.hadoop.conf.Configuration object
      jobject jConfiguration =
          constructNewObjectOfClass(env, HADOOP_CONF, "()V");
      if (jConfiguration == NULL) {
          fprintf(stderr, "Can't construct instance of class "
                  "org.apache.hadoop.conf.Configuration\n");
          errno = EINTERNAL;
          destroyLocalReference(env, jSrcPath);
          destroyLocalReference(env, jDstPath);
          return -1;
      }
  
      //FileUtil::copy
      jboolean deleteSource = 0; //Only copy
      jvalue jVal;
      if (invokeMethod(env, &jVal, STATIC, 
                       NULL, "org/apache/hadoop/fs/FileUtil", "copy",
                       "(Lorg/apache/hadoop/fs/FileSystem;Lorg/apache/hadoop/fs/Path;Lorg/apache/hadoop/fs/FileSystem;Lorg/apache/hadoop/fs/Path;ZLorg/apache/hadoop/conf/Configuration;)Z",
                       jSrcFS, jSrcPath, jDstFS, jDstPath, deleteSource, 
                       jConfiguration) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileUtil::copy failed!\n");
          errno = EINTERNAL;
          retval = -1;
          goto done;
      }
  
      done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jConfiguration);
      destroyLocalReference(env, jSrcPath);
      destroyLocalReference(env, jDstPath);
    
      return retval;
  }
  
  
  
  int hdfsMove(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst)
  {
      //JAVA EQUIVALENT
      //  FileUtil::copy(srcFS, srcPath, dstFS, dstPath,
      //                 deleteSource = true, conf)
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      //Parameters
      jobject jSrcFS = (jobject)srcFS;
      jobject jDstFS = (jobject)dstFS;
  
      jobject jSrcPath = NULL;
      jobject jDstPath = NULL;
  
      jSrcPath = constructNewObjectOfPath(env, src);
      if (jSrcPath == NULL) {
          return -1;
      }
  
      jDstPath = constructNewObjectOfPath(env, dst);
      if (jDstPath == NULL) {
          destroyLocalReference(env, jSrcPath);
          return -1;
      }
  
      int retval = 0;
  
      //Create the org.apache.hadoop.conf.Configuration object
      jobject jConfiguration =
          constructNewObjectOfClass(env, HADOOP_CONF, "()V");
      if (jConfiguration == NULL) {
          fprintf(stderr, "Can't construct instance of class "
                  "org.apache.hadoop.conf.Configuration\n");
          errno = EINTERNAL;
          destroyLocalReference(env, jSrcPath);
          destroyLocalReference(env, jDstPath);
          return -1;
      }
  
      //FileUtil::copy
      jboolean deleteSource = 1; //Delete src after copy
      jvalue jVal;
      if (invokeMethod(env, &jVal, STATIC, NULL,
                       "org/apache/hadoop/fs/FileUtil", "copy",
                  "(Lorg/apache/hadoop/fs/FileSystem;Lorg/apache/hadoop/fs/Path;Lorg/apache/hadoop/fs/FileSystem;Lorg/apache/hadoop/fs/Path;ZLorg/apache/hadoop/conf/Configuration;)Z",
                       jSrcFS, jSrcPath, jDstFS, jDstPath, deleteSource, 
                       jConfiguration) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileUtil::copy(move) failed!\n");
          errno = EINTERNAL;
          retval = -1;
          goto done;
      }
  
      done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jConfiguration);
      destroyLocalReference(env, jSrcPath);
      destroyLocalReference(env, jDstPath);
    
      return retval;
  }
  
  
  
  int hdfsDelete(hdfsFS fs, const char* path)
  {
      // JAVA EQUIVALENT:
      //  File f = new File(path);
      //  bool retval = fs.delete(f);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //Create an object of java.io.File
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return -1;
      }
  
      //Delete the file
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "delete", "(Lorg/apache/hadoop/fs/Path;)Z",
                       jPath) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::delete failed!\n");
          errno = EINTERNAL;
          return -1;
      }
  
      //Delete unnecessary local references
      destroyLocalReference(env, jPath);
  
      return (jVal.z) ? 0 : -1;
  }
  
  
  
  int hdfsRename(hdfsFS fs, const char* oldPath, const char* newPath)
  {
      // JAVA EQUIVALENT:
      //  Path old = new Path(oldPath);
      //  Path new = new Path(newPath);
      //  fs.rename(old, new);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //Create objects of org.apache.hadoop.fs.Path
      jobject jOldPath = NULL;
      jobject jNewPath = NULL;
  
      jOldPath = constructNewObjectOfPath(env, oldPath);
      if (jOldPath == NULL) {
          return -1;
      }
  
      jNewPath = constructNewObjectOfPath(env, newPath);
      if (jNewPath == NULL) {
          destroyLocalReference(env, jOldPath);
          return -1;
      }
  
      //Rename the file
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS, "rename",
                       JMETHOD2(JPARAM(HADOOP_PATH), JPARAM(HADOOP_PATH), "Z"),
                       jOldPath, jNewPath) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::rename failed!\n");
          errno = EINTERNAL;
          return -1;
      }
  
      //Delete unnecessary local references
      destroyLocalReference(env, jOldPath);
      destroyLocalReference(env, jNewPath);
  
      return (jVal.z) ? 0 : -1;
  }
  
  
  
  char* hdfsGetWorkingDirectory(hdfsFS fs, char* buffer, size_t bufferSize)
  {
      // JAVA EQUIVALENT:
      //  Path p = fs.getWorkingDirectory(); 
      //  return p.toString()
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
      jobject jPath = NULL;
      jvalue jVal;
  
      //FileSystem::getWorkingDirectory()
      if (invokeMethod(env, &jVal, INSTANCE, jFS,
                       HADOOP_FS, "getWorkingDirectory",
                       "()Lorg/apache/hadoop/fs/Path;") != 0 ||
          jVal.l == NULL) {
          fprintf(stderr, "Call to FileSystem::getWorkingDirectory failed!\n");
          errno = EINTERNAL;
          return NULL;
      }
      jPath = jVal.l;
  
      //Path::toString()
      jstring jPathString;
      if (invokeMethod(env, &jVal, INSTANCE, jPath, 
                       "org/apache/hadoop/fs/Path", "toString",
                       "()Ljava/lang/String;") != 0) { 
          fprintf(stderr, "Call to Path::toString failed!\n");
          errno = EINTERNAL;
          destroyLocalReference(env, jPath);
          return NULL;
      }
      jPathString = jVal.l;
  
      const char *jPathChars = (const char*)
          ((*env)->GetStringUTFChars(env, jPathString, NULL));
  
      //Copy to user-provided buffer
      strncpy(buffer, jPathChars, bufferSize);
  
      //Delete unnecessary local references
      (*env)->ReleaseStringUTFChars(env, jPathString, jPathChars);
  
      destroyLocalReference(env, jPathString);
      destroyLocalReference(env, jPath);
  
      return buffer;
  }
  
  
  
  int hdfsSetWorkingDirectory(hdfsFS fs, const char* path)
  {
      // JAVA EQUIVALENT:
      //  fs.setWorkingDirectory(Path(path)); 
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
      int retval = 0;
  
      //Create an object of org.apache.hadoop.fs.Path
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return -1;
      }
  
      //FileSystem::setWorkingDirectory()
      if (invokeMethod(env, NULL, INSTANCE, jFS, HADOOP_FS,
                       "setWorkingDirectory", 
                       "(Lorg/apache/hadoop/fs/Path;)V", jPath) != 0) {
          fprintf(stderr, "Call to FileSystem::setWorkingDirectory failed!\n");
          errno = EINTERNAL;
          retval = -1;
      }
  
      //Delete unnecessary local references
      destroyLocalReference(env, jPath);
  
      return retval;
  }
  
  
  
  int hdfsCreateDirectory(hdfsFS fs, const char* path)
  {
      // JAVA EQUIVALENT:
      //  fs.mkdirs(new Path(path));
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //Create an object of org.apache.hadoop.fs.Path
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return -1;
      }
  
      //Create the directory
      jvalue jVal;
      jVal.z = 0;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "mkdirs", "(Lorg/apache/hadoop/fs/Path;)Z",
                       jPath) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs.FileSystem::"
                  "mkdirs failed!\n");
          errno = EINTERNAL;
          goto done;
      }
  
   done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jPath);
  
      return (jVal.z) ? 0 : -1;
  }
  
  
  int hdfsSetReplication(hdfsFS fs, const char* path, int16_t replication)
  {
      // JAVA EQUIVALENT:
      //  fs.setReplication(new Path(path), replication);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //Create an object of org.apache.hadoop.fs.Path
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return -1;
      }
  
      //Create the directory
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "setReplication", "(Lorg/apache/hadoop/fs/Path;S)Z",
                       jPath, replication) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs.FileSystem::"
                  "setReplication failed!\n");
          errno = EINTERNAL;
          goto done;
      }
  
   done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jPath);
  
      return (jVal.z) ? 0 : -1;
  }
  
  
  
  char***
  hdfsGetHosts(hdfsFS fs, const char* path, tOffset start, tOffset length)
  {
      // JAVA EQUIVALENT:
      //  fs.getFileBlockLoctions(new Path(path), start, length);
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //Create an object of org.apache.hadoop.fs.Path
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return NULL;
      }
  
      //org.apache.hadoop.fs.FileSystem::getFileBlockLocations
      char*** blockHosts = NULL;
      jobjectArray jBlockLocations;;
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS,
                       HADOOP_FS, "getFileBlockLocations", 
                       "(Lorg/apache/hadoop/fs/Path;JJ)"
                       "[Lorg/apache/hadoop/fs/BlockLocation;",
                       jPath, start, length) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::getFileBlockLocations failed!\n");
          errno = EINTERNAL;
          destroyLocalReference(env, jPath);
          return NULL;
      }
      jBlockLocations = jVal.l;
  
      //Figure out no of entries in jBlockLocations
      //Allocate memory and add NULL at the end
      jsize jNumFileBlocks = (*env)->GetArrayLength(env, jBlockLocations);
  
      blockHosts = malloc(sizeof(char**) * (jNumFileBlocks+1));
      if (blockHosts == NULL) {
          errno = ENOMEM;
          goto done;
      }
      blockHosts[jNumFileBlocks] = NULL;
      if (jNumFileBlocks == 0) {
          errno = 0;
          goto done;
      }
  
      //Now parse each block to get hostnames
      int i = 0;
      for (i=0; i < jNumFileBlocks; ++i) {
          jobject jFileBlock =
              (*env)->GetObjectArrayElement(env, jBlockLocations, i);
          
          jvalue jVal;
          jobjectArray jFileBlockHosts;
          if (invokeMethod(env, &jVal, INSTANCE, jFileBlock, HADOOP_BLK_LOC,
                           "getHosts", "()[Ljava/lang/String;") ||
                  jVal.l == NULL) {
              fprintf(stderr, "Call to org.apache.hadoop.fs.BlockLocation::"
                      "getHosts failed!\n");
              errno = EINTERNAL;
              destroyLocalReference(env, jPath);
              destroyLocalReference(env, jBlockLocations);
              return NULL;
          }
          
          jFileBlockHosts = jVal.l;
          //Figure out no of hosts in jFileBlockHosts
          //Allocate memory and add NULL at the end
          jsize jNumBlockHosts = (*env)->GetArrayLength(env, jFileBlockHosts);
          blockHosts[i] = malloc(sizeof(char*) * (jNumBlockHosts+1));
          if (blockHosts[i] == NULL) {
              int x = 0;
              for (x=0; x < i; ++x) {
                  free(blockHosts[x]);
              }
              free(blockHosts);
              errno = ENOMEM;
              goto done;
          }
          blockHosts[i][jNumBlockHosts] = NULL;
  
          //Now parse each hostname
          int j = 0;
          const char *hostName;
          for (j=0; j < jNumBlockHosts; ++j) {
              jstring jHost =
                  (*env)->GetObjectArrayElement(env, jFileBlockHosts, j);
             
              hostName =
                  (const char*)((*env)->GetStringUTFChars(env, jHost, NULL));
              blockHosts[i][j] = strdup(hostName);
  
              (*env)->ReleaseStringUTFChars(env, jHost, hostName);
              destroyLocalReference(env, jHost);
          }
  
          destroyLocalReference(env, jFileBlockHosts);
      }
    
      done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jPath);
      destroyLocalReference(env, jBlockLocations);
  
      return blockHosts;
  }
  
  
  void hdfsFreeHosts(char ***blockHosts)
  {
      int i, j;
      for (i=0; blockHosts[i]; i++) {
          for (j=0; blockHosts[i][j]; j++) {
              free(blockHosts[i][j]);
          }
          free(blockHosts[i]);
      }
      free(blockHosts);
  }
  
  
  tOffset hdfsGetDefaultBlockSize(hdfsFS fs)
  {
      // JAVA EQUIVALENT:
      //  fs.getDefaultBlockSize();
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //FileSystem::getDefaultBlockSize()
      tOffset blockSize = -1;
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "getDefaultBlockSize", "()J") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::getDefaultBlockSize failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      blockSize = jVal.j;
  
      return blockSize;
  }
  
  
  
  tOffset hdfsGetCapacity(hdfsFS fs)
  {
      // JAVA EQUIVALENT:
      //  fs.getRawCapacity();
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      if (!((*env)->IsInstanceOf(env, jFS, 
                                 globalClassReference(HADOOP_DFS, env)))) {
          fprintf(stderr, "hdfsGetCapacity works only on a "
                  "DistributedFileSystem!\n");
          return -1;
      }
  
      //FileSystem::getRawCapacity()
      jvalue  jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_DFS,
                       "getRawCapacity", "()J") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::getRawCapacity failed!\n");
          errno = EINTERNAL;
          return -1;
      }
  
      return jVal.j;
  }
  
  
    
  tOffset hdfsGetUsed(hdfsFS fs)
  {
      // JAVA EQUIVALENT:
      //  fs.getRawUsed();
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      if (!((*env)->IsInstanceOf(env, jFS, 
                                 globalClassReference(HADOOP_DFS, env)))) {
          fprintf(stderr, "hdfsGetUsed works only on a "
                  "DistributedFileSystem!\n");
          return -1;
      }
  
      //FileSystem::getRawUsed()
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_DFS,
                       "getRawUsed", "()J") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::getRawUsed failed!\n");
          errno = EINTERNAL;
          return -1;
      }
  
      return jVal.j;
  }
  
  
   
  static int
  getFileInfoFromStat(JNIEnv *env, jobject jStat, hdfsFileInfo *fileInfo)
  {
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jStat,
                       HADOOP_STAT, "isDir", "()Z") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileStatus::isDir failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      fileInfo->mKind = jVal.z ? kObjectKindDirectory : kObjectKindFile;
  
      if (invokeMethod(env, &jVal, INSTANCE, jStat,
                       HADOOP_STAT, "getReplication", "()S") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileStatus::getReplication failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      fileInfo->mReplication = jVal.s;
  
      if (invokeMethod(env, &jVal, INSTANCE, jStat,
                       HADOOP_STAT, "getBlockSize", "()J") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileStatus::getBlockSize failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      fileInfo->mBlockSize = jVal.j;
  
      if (invokeMethod(env, &jVal, INSTANCE, jStat,
                       HADOOP_STAT, "getModificationTime", "()J") != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileStatus::getModificationTime failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      fileInfo->mLastMod = (tTime) (jVal.j / 1000);
  
      if (fileInfo->mKind == kObjectKindFile) {
          if (invokeMethod(env, &jVal, INSTANCE, jStat,
                           HADOOP_STAT, "getLen", "()J") != 0) {
              fprintf(stderr, "Call to org.apache.hadoop.fs."
                      "FileStatus::getLen failed!\n");
              errno = EINTERNAL;
              return -1;
          }
          fileInfo->mSize = jVal.j;
      }
  
      jobject jPath;
      if (invokeMethod(env, &jVal, INSTANCE, jStat, HADOOP_STAT,
                       "getPath", "()Lorg/apache/hadoop/fs/Path;") ||
              jVal.l == NULL) { 
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileStatus::getPath failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      jPath = jVal.l;
  
      jstring     jPathName;
      const char *cPathName;
      if (invokeMethod(env, &jVal, INSTANCE, jPath, HADOOP_PATH,
                       "toString", "()Ljava/lang/String;")) { 
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "Path::toString failed!\n");
          errno = EINTERNAL;
          destroyLocalReference(env, jPath);
          return -1;
      }
      jPathName = jVal.l;
      cPathName = (const char*) ((*env)->GetStringUTFChars(env, jPathName, NULL));
      fileInfo->mName = strdup(cPathName);
      (*env)->ReleaseStringUTFChars(env, jPathName, cPathName);
      destroyLocalReference(env, jPath);
      destroyLocalReference(env, jPathName);
  
      return 0;
  }
  
  static int
  getFileInfo(JNIEnv *env, jobject jFS, jobject jPath, hdfsFileInfo *fileInfo)
  {
      // JAVA EQUIVALENT:
      //  fs.isDirectory(f)
      //  fs.lastModified() ??
      //  fs.getLength(f)
      //  f.getPath()
  
      jobject jStat;
      jvalue  jVal;
  
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "exists", JMETHOD1(JPARAM(HADOOP_PATH), "Z"),
                       jPath) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::exists failed!\n");
          errno = EINTERNAL;
          return -1;
      }
  
      if (jVal.z == 0) {
        errno = ENOENT;
        return -1;
      }
  
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_FS,
                       "getFileStatus", JMETHOD1(JPARAM(HADOOP_PATH), JPARAM(HADOOP_STAT)),
                       jPath) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::getFileStatus failed!\n");
          errno = EINTERNAL;
          return -1;
      }
      jStat = jVal.l;
      int ret =  getFileInfoFromStat(env, jStat, fileInfo); 
      destroyLocalReference(env, jStat);
      return ret;
  }
  
  
  
  hdfsFileInfo* hdfsListDirectory(hdfsFS fs, const char* path, int *numEntries)
  {
      // JAVA EQUIVALENT:
      //  Path p(path);
      //  Path []pathList = fs.listPaths(p)
      //  foreach path in pathList 
      //    getFileInfo(path)
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //Create an object of org.apache.hadoop.fs.Path
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return NULL;
      }
  
      hdfsFileInfo *pathList = 0; 
  
      jobjectArray jPathList = NULL;
      jvalue jVal;
      if (invokeMethod(env, &jVal, INSTANCE, jFS, HADOOP_DFS, "listStatus",
                       JMETHOD1(JPARAM(HADOOP_PATH), JARRPARAM(HADOOP_STAT)),
                       jPath) != 0) {
          fprintf(stderr, "Call to org.apache.hadoop.fs."
                  "FileSystem::listStatus failed!\n");
          errno = EINTERNAL;
          destroyLocalReference(env, jPath);
          return NULL;
      }
      jPathList = jVal.l;
  
      //Figure out no of entries in that directory
      jsize jPathListSize = (*env)->GetArrayLength(env, jPathList);
      *numEntries = jPathListSize;
      if (jPathListSize == 0) {
          errno = 0;
          goto done;
      }
  
      //Allocate memory
      pathList = calloc(jPathListSize, sizeof(hdfsFileInfo));
      if (pathList == NULL) {
          errno = ENOMEM;
          goto done;
      }
  
      //Save path information in pathList
      jsize i;
      jobject tmpStat;
      for (i=0; i < jPathListSize; ++i) {
          tmpStat = (*env)->GetObjectArrayElement(env, jPathList, i);
          if (getFileInfoFromStat(env, tmpStat, &pathList[i])) {
              errno = EINTERNAL;
              hdfsFreeFileInfo(pathList, jPathListSize);
              destroyLocalReference(env, tmpStat);
              pathList = NULL;
              goto done;
          }
          destroyLocalReference(env, tmpStat);
      }
  
      done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jPath);
      destroyLocalReference(env, jPathList);
  
      return pathList;
  }
  
  
  
  hdfsFileInfo *hdfsGetPathInfo(hdfsFS fs, const char* path)
  {
      // JAVA EQUIVALENT:
      //  File f(path);
      //  fs.isDirectory(f)
      //  fs.lastModified() ??
      //  fs.getLength(f)
      //  f.getPath()
  
      //Get the JNIEnv* corresponding to current thread
      JNIEnv* env = getJNIEnv();
  
      jobject jFS = (jobject)fs;
  
      //Create an object of org.apache.hadoop.fs.Path
      jobject jPath = constructNewObjectOfPath(env, path);
      if (jPath == NULL) {
          return NULL;
      }
  
      hdfsFileInfo *fileInfo = calloc(1, sizeof(hdfsFileInfo));
      if (getFileInfo(env, jFS, jPath, fileInfo)) {
          hdfsFreeFileInfo(fileInfo, 1);
          fileInfo = NULL;
          goto done;
      }
  
      done:
  
      //Delete unnecessary local references
      destroyLocalReference(env, jPath);
  
      return fileInfo;
  }
  
  
  
  void hdfsFreeFileInfo(hdfsFileInfo *hdfsFileInfo, int numEntries)
  {
      //Free the mName
      int i;
      for (i=0; i < numEntries; ++i) {
          if (hdfsFileInfo[i].mName) {
              free(hdfsFileInfo[i].mName);
          }
      }
  
      //Free entire block
      free(hdfsFileInfo);
  }
  
  
  
  
  /**
   * vim: ts=4: sw=4: et:
   */
  