feat: support ssl cert

Author: TheoRadig

.gitignore

@@ -1,3 +1,8 @@
+# Build directories of CLion.
+cmake-build-*
+
+.idea
+
 # Prerequisites
 *.d
 

include/webhdfspp/webhdfspp.h

@@ -21,6 +21,7 @@
 #include "status.h"
 
 #include <functional>
+#include <memory>
 #include <string>
 #include <utility>
 #include <vector>
@@ -34,9 +35,16 @@ class non_copyable {
   void operator=(non_copyable const &) = delete;
 };
 
+struct Options {
+    std::vector<std::pair<std::string, short>> namenodes;
+    char* ssl_cert;
+    char* ssl_key;
+    std::string scheme;
+};
+
 class IoService : non_copyable {
 public:
-  static IoService *New();
+  static IoService *New(const Options &options);
   virtual Status Run() = 0;
   virtual void Stop() = 0;
   virtual ~IoService();
@@ -45,10 +53,6 @@ class IoService : non_copyable {
 class InputStream;
 class OutputStream;
 
-struct Options {
-  std::vector<std::pair<std::string, short>> namenodes;
-};
-
 struct FileStatus {
   enum FileType {
     FILE,

lib/webhdfspp/filesystem.cc

@@ -66,8 +66,10 @@ WebHdfsFileSystem::WebHdfsFileSystem(const Options &options,
 
 Status WebHdfsFileSystem::Delete(const std::string &path, bool recursive) {
   const auto &nn = options_.namenodes[active_endpoint_];
-  URIBuilder builder;
-  auto uri = builder.Scheme("http")
+  const auto &scheme = options_.scheme;
+
+    URIBuilder builder;
+  auto uri = builder.Scheme(scheme)
                  .Host(nn.first)
                  .Port(nn.second)
                  .Path("/webhdfs/v1" + path)
@@ -88,8 +90,10 @@ Status WebHdfsFileSystem::Delete(const std::string &path, bool recursive) {
 Status WebHdfsFileSystem::GetFileStatus(const std::string &path,
                                         FileStatus *stat) {
   const auto &nn = options_.namenodes[active_endpoint_];
-  URIBuilder builder;
-  auto uri = builder.Scheme("http")
+  const auto &scheme = options_.scheme;
+
+    URIBuilder builder;
+  auto uri = builder.Scheme(scheme)
                  .Host(nn.first)
                  .Port(nn.second)
                  .Path("/webhdfs/v1" + path)
@@ -142,8 +146,7 @@ Status WebHdfsFileSystem::Exists(const std::string &path, bool *result) {
 
 Status WebHdfsFileSystem::Open(const std::string &path,
                                std::unique_ptr<InputStream> *is) {
-  const auto &nn = options_.namenodes[active_endpoint_];
-  auto is_ptr = new InputStreamImpl(nn, path, io_service_);
+  auto is_ptr = new InputStreamImpl(options_, path, io_service_, active_endpoint_);
   is->reset(is_ptr);
   return Status::OK();
 }

lib/webhdfspp/inputstream.cc

@@ -22,18 +22,19 @@ namespace webhdfspp {
 
 InputStream::~InputStream() {}
 
-InputStreamImpl::InputStreamImpl(const std::pair<std::string, short> &nn,
+InputStreamImpl::InputStreamImpl(const Options &options,
                                  const std::string &path,
-                                 std::shared_ptr<IoServiceImpl> io_service)
-    : nn_(nn), path_(path), io_service_(io_service) {}
+                                 std::shared_ptr<IoServiceImpl> io_service, int active_endpoint)
+    : options_(options), path_(path), io_service_(io_service), active_endpoint_(active_endpoint) {}
 
 Status InputStreamImpl::PositionRead(
     size_t max_read_bytes, size_t offset,
     const std::function<size_t(const char *, size_t)> &on_data_arrived) {
+    const auto nn = options_.namenodes[active_endpoint_];
   URIBuilder builder;
-  auto uri = builder.Scheme("http")
-                 .Host(nn_.first)
-                 .Port(nn_.second)
+  auto uri = builder.Scheme(options_.scheme)
+                 .Host(nn.first)
+                 .Port(nn.second)
                  .Path("/webhdfs/v1" + path_)
                  .Param("op", "OPEN");
 

lib/webhdfspp/inputstream_impl.h

@@ -25,17 +25,18 @@ namespace webhdfspp {
 
 class InputStreamImpl : public InputStream {
 public:
-  InputStreamImpl(const std::pair<std::string, short> &nn,
+  InputStreamImpl(const Options &options,
                   const std::string &path,
-                  std::shared_ptr<IoServiceImpl> io_service);
+                  std::shared_ptr<IoServiceImpl> io_service, int active_endpoint);
   virtual Status PositionRead(size_t max_read_bytes, size_t offset,
                               const std::function<size_t(const char *, size_t)>
                                   &on_data_arrived) override;
 
 private:
-  std::pair<std::string, short> nn_;
+  const Options options_;
   std::string path_;
   std::shared_ptr<IoServiceImpl> io_service_;
+  int active_endpoint_;
 };
 
 } // namespace webhdfspp

lib/webhdfspp/io_service.cc

@@ -23,11 +23,11 @@
 
 namespace webhdfspp {
 
-IoServiceImpl::IoServiceImpl() {}
+IoServiceImpl::IoServiceImpl(const Options &options) : options_(options) {}
 
 IoServiceImpl::~IoServiceImpl() {}
 
-IoService *IoService::New() { return new IoServiceImpl(); }
+IoService *IoService::New(const Options &options) { return new IoServiceImpl(options); }
 
 IoService::~IoService() {}
 
@@ -44,6 +44,11 @@ Status IoServiceImpl::DoNNRequest(const URIBuilder &uri,
   char error_buffer[CURL_ERROR_SIZE];
   auto uri_str = uri.Build();
 
+  if (options_.ssl_cert && options_.ssl_key) {
+      curl_easy_setopt(handle, CURLOPT_SSLCERT, options_.ssl_cert);
+      curl_easy_setopt(handle, CURLOPT_SSLKEY, options_.ssl_key);
+  }
+
   curl_easy_setopt(handle, CURLOPT_URL, uri_str.c_str());
   curl_easy_setopt(handle, CURLOPT_ERRORBUFFER, error_buffer);
   curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, method.c_str());

lib/webhdfspp/io_service_impl.h

@@ -29,7 +29,7 @@ namespace webhdfspp {
 class IoServiceImpl : public IoService,
                       public std::enable_shared_from_this<IoServiceImpl> {
 public:
-  IoServiceImpl();
+  IoServiceImpl(const Options &options);
   ~IoServiceImpl();
   virtual Status Run() override;
   virtual void Stop() override;
@@ -46,6 +46,8 @@ class IoServiceImpl : public IoService,
   static CURLcode
   DNGetCallback(char *in, size_t size, size_t nmemb,
                 std::function<size_t(const char *, size_t)> *on_data_arrived);
+
+  const Options options_;
 };
 } // namespace webhdfspp