Hdfs 9144 merge

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/libhdfspp/hdfs.h

@@ -20,6 +20,7 @@
 
 #include "libhdfspp/options.h"
 #include "libhdfspp/status.h"
+#include "common/cancelable.h"
 
 #include <functional>
 #include <set>
@@ -54,9 +55,9 @@ class IoService {
 };
 
 /**
- * Applications opens an InputStream to read files in HDFS.
+ * Applications opens a FileHandle to read files in HDFS.
  **/
-class InputStream {
+class FileHandle {
 public:
   /**
    * Read data from a specific position. The current implementation
@@ -71,12 +72,13 @@ class InputStream {
    * The handler returns the datanode that serves the block and the number of
    * bytes has read.
    **/
-  virtual void
+  virtual CancelHandle
   PositionRead(void *buf, size_t nbyte, uint64_t offset,
-               const std::set<std::string> &excluded_datanodes,
-               const std::function<void(const Status &, const std::string &,
-                                        size_t)> &handler) = 0;
-  virtual ~InputStream();
+               const std::function<void(const Status &, size_t)> &handler) = 0;
+
+  virtual size_t PositionRead(void *buf, size_t nbyte, off_t offset) = 0;
+
+  virtual ~FileHandle();
 };
 
 /**
@@ -93,15 +95,24 @@ class FileSystem {
   New(IoService *io_service, const Options &options, const std::string &server,
       const std::string &service,
       const std::function<void(const Status &, FileSystem *)> &handler);
+
+  /* Synchronous call of New*/
+  static FileSystem *
+  New(IoService *io_service, const Options &options, const std::string &server,
+      const std::string &service);
+
   /**
    * Open a file on HDFS. The call issues an RPC to the NameNode to
    * gather the locations of all blocks in the file and to return a
    * new instance of the @ref InputStream object.
    **/
   virtual void
   Open(const std::string &path,
-       const std::function<void(const Status &, InputStream *)> &handler) = 0;
-  virtual ~FileSystem();
+       const std::function<void(const Status &, FileHandle *)> &handler) = 0;
+  virtual Status Open(const std::string &path, FileHandle **handle);
+
+  virtual ~FileSystem() {};
+
 };
 }
 

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/CMakeLists.txt

@@ -21,4 +21,5 @@ add_subdirectory(fs)
 add_subdirectory(reader)
 add_subdirectory(rpc)
 add_subdirectory(proto)
+add_subdirectory(connection)
 add_subdirectory(bindings)

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/CMakeLists.txt

@@ -16,5 +16,5 @@
 # under the License.
 
 
-add_library(bindings_c hdfs.cc hdfs_cpp.cc)
+add_library(bindings_c hdfs.cc)
 add_dependencies(bindings_c fs rpc reader proto common fs rpc reader proto common)

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc

@@ -16,24 +16,26 @@
  * limitations under the License.
  */
 
-#include "hdfs_cpp.h"
+#include "fs/filesystem.h"
 
+#include <hdfs/hdfs.h>
+#include <string>
 #include <cstring>
 #include <iostream>
 
 using namespace hdfs;
 
 /* Seperate the handles used by the C api from the C++ API*/
 struct hdfs_internal {
-  hdfs_internal(HadoopFileSystem *p) : filesystem_(p) {}
-  hdfs_internal(std::unique_ptr<HadoopFileSystem> p)
+  hdfs_internal(FileSystem *p) : filesystem_(p) {}
+  hdfs_internal(std::unique_ptr<FileSystem> p)
       : filesystem_(std::move(p)) {}
   virtual ~hdfs_internal(){};
-  HadoopFileSystem *get_impl() { return filesystem_.get(); }
-  const HadoopFileSystem *get_impl() const { return filesystem_.get(); }
+  FileSystem *get_impl() { return filesystem_.get(); }
+  const FileSystem *get_impl() const { return filesystem_.get(); }
 
  private:
-  std::unique_ptr<HadoopFileSystem> filesystem_;
+  std::unique_ptr<FileSystem> filesystem_;
 };
 
 struct hdfsFile_internal {
@@ -65,17 +67,17 @@ static void ReportError(int errnum, std::string msg) {
 int hdfsFileIsOpenForRead(hdfsFile file) {
   /* files can only be open for reads at the moment, do a quick check */
   if (file) {
-    return file->get_impl()->IsOpenForRead();
+    return true; // Update implementation when we get file writing
   }
   return false;
 }
 
 hdfsFS hdfsConnect(const char *nn, tPort port) {
-  HadoopFileSystem *fs = new HadoopFileSystem();
-  Status stat = fs->Connect(nn, port);
-  if (!stat.ok()) {
+  std::string port_as_string = std::to_string(port);
+  IoService * io_service = IoService::New();
+  FileSystem *fs = FileSystem::New(io_service, Options(), nn, port_as_string);
+  if (!fs) {
     ReportError(ENODEV, "Unable to connect to NameNode.");
-    delete fs;
     return nullptr;
   }
   return new hdfs_internal(fs);
@@ -102,7 +104,7 @@ hdfsFile hdfsOpenFile(hdfsFS fs, const char *path, int flags, int bufferSize,
     return nullptr;
   }
   FileHandle *f = nullptr;
-  Status stat = fs->get_impl()->OpenFileForRead(path, &f);
+  Status stat = fs->get_impl()->Open(path, &f);
   if (!stat.ok()) {
     return nullptr;
   }
@@ -133,5 +135,5 @@ tSize hdfsPread(hdfsFS fs, hdfsFile file, tOffset position, void *buffer,
     return -1;
   }
 
-  return file->get_impl()->Pread(buffer, length, position);
+  return file->get_impl()->PositionRead(buffer, length, position);
 }