address comments

vesoft-inc · Dec 6, 2021 · 0148a8c · 0148a8c
1 parent 9b81ef7
commit 0148a8c
Show file tree

Hide file tree

Showing 22 changed files with 89 additions and 94 deletions.
diff --git a/include/nebula/mclient/MConfig.h b/include/nebula/mclient/MConfig.h
@@ -0,0 +1,20 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+namespace nebula {
+
+struct MConfig {
+  int32_t connTimeoutInMs_{1000};         // It's as same as FLAGS_conn_timeout_ms in nebula
+  int32_t clientTimeoutInMs_{60 * 1000};  // It's as same as FLAG_meta_client_timeout_ms in nebula
+  bool enableSSL_{false};
+  std::string CAPath_;
+};
+
+}  // namespace nebula
diff --git a/include/nebula/mclient/MetaClient.h b/include/nebula/mclient/MetaClient.h
@@ -14,7 +14,7 @@
 
 #include "common/datatypes/HostAddr.h"
 #include "common/thrift/ThriftTypes.h"
-#include "nebula/sclient/ScanEdgeIter.h"
+#include "nebula/mclient/MConfig.h"
 
 struct pair_hash {
   template <class T1, class T2>
@@ -63,7 +63,8 @@ using SpaceEdgeNameTypeMap =
 
 class MetaClient {
  public:
-  explicit MetaClient(const std::vector<std::string> &metaAddrs);
+  explicit MetaClient(const std::vector<std::string> &metaAddrs,
+                      const MConfig &mConfig = MConfig{});
 
   ~MetaClient();
 
@@ -103,6 +104,7 @@ class MetaClient {
 
  private:
   std::vector<HostAddr> metaAddrs_;
+  MConfig mConfig_;
   SpaceNameIdMap spaceIndexByName_;
   SpaceEdgeNameTypeMap spaceEdgeIndexByName_;
   std::unordered_map<std::pair<GraphSpaceID, PartitionID>, HostAddr, pair_hash> spacePartLeaderMap_;

diff --git a/include/nebula/sclient/Init.h b/include/nebula/sclient/Init.h
diff --git a/include/nebula/sclient/SConfig.h b/include/nebula/sclient/SConfig.h
@@ -0,0 +1,21 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+namespace nebula {
+
+struct SConfig {
+  int32_t connTimeoutInMs_{1000};  // in ms. It's as same as FLAGS_conn_timeout_ms in nebula
+  int32_t clientTimeoutInMs_{60 *
+                             1000};  // in ms. It's as same as FLAG_meta_client_timeout_ms in nebula
+  bool enableSSL_{false};
+  std::string CAPath_;
+};
+
+}  // namespace nebula
diff --git a/include/nebula/sclient/ScanEdgeIter.h b/include/nebula/sclient/ScanEdgeIter.h
@@ -32,7 +32,6 @@ struct ScanEdgeIter {
   storage::cpp2::ScanEdgeRequest* req_;
   bool hasNext_;
   std::string nextCursor_;
-  bool firstScan_{true};
 };
 
 }  // namespace nebula
diff --git a/include/nebula/sclient/StorageClient.h b/include/nebula/sclient/StorageClient.h
@@ -15,6 +15,7 @@
 #include "common/datatypes/HostAddr.h"
 #include "common/thrift/ThriftTypes.h"
 #include "nebula/mclient/MetaClient.h"
+#include "nebula/sclient/SConfig.h"
 #include "nebula/sclient/ScanEdgeIter.h"
 
 namespace folly {
@@ -51,15 +52,17 @@ class ScanEdgeResponse;
 }  // namespace cpp2
 }  // namespace storage
 
-#define DEFAULT_LIMIT 1000
+#define DEFAULT_LIMIT std::numeric_limits<int64_t>::max()
 #define DEFAULT_START_TIME 0
 #define DEFAULT_END_TIME std::numeric_limits<int64_t>::max()
 
 class StorageClient {
   friend struct ScanEdgeIter;
 
  public:
-  explicit StorageClient(const std::vector<std::string>& metaAddrs);
+  explicit StorageClient(const std::vector<std::string>& metaAddrs,
+                         const MConfig& mConfig = MConfig{},
+                         const SConfig& sConfig = SConfig{});
 
   ~StorageClient();
 
@@ -88,6 +91,7 @@ class StorageClient {
                    folly::Promise<std::pair<bool, Response>> pro);
 
   std::unique_ptr<MetaClient> mClient_;
+  SConfig sConfig_;
   std::shared_ptr<folly::IOThreadPoolExecutor> ioExecutor_;
   std::shared_ptr<thrift::ThriftClientManager<storage::cpp2::GraphStorageServiceAsyncClient>>
       clientsMan_;

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
@@ -35,13 +35,13 @@ set(NEBULA_COMMON_SOURCES
     geo/io/wkb/WKBWriter.cpp
     geo/io/wkb/ByteOrderDataIOStream.cpp
     Init.cpp
+    SSLConfig.cpp
 )
 
 set(NEBULA_CLIENT_SOURCES
     client/Connection.cpp
     client/ConnectionPool.cpp
     client/Session.cpp
-    client/SSLConfig.cpp
 )
 
 set(NEBULA_MCLIENT_SOURCES

diff --git a/src/ClientTest.h b/src/ClientTest.h
diff --git a/src/client/SSLConfig.cpp → src/SSLConfig.cpp b/src/client/SSLConfig.cpp → src/SSLConfig.cpp
diff --git a/src/client/SSLConfig.h → src/SSLConfig.h b/src/client/SSLConfig.h → src/SSLConfig.h
diff --git a/src/client/Connection.cpp b/src/client/Connection.cpp
@@ -14,7 +14,7 @@
 #include <memory>
 #include <stdexcept>
 
-#include "./SSLConfig.h"
+#include "../SSLConfig.h"
 #include "interface/gen-cpp2/GraphServiceAsyncClient.h"
 
 namespace nebula {

diff --git a/src/client/tests/AddressTest.cpp b/src/client/tests/AddressTest.cpp
@@ -10,7 +10,7 @@
 #include <nebula/client/ConnectionPool.h>
 #include <nebula/client/Session.h>
 
-#include "../../ClientTest.h"
+#include "./ClientTest.h"
 
 // Require a nebula server could access
 

diff --git a/src/client/tests/ConfigTest.cpp b/src/client/tests/ConfigTest.cpp
@@ -10,7 +10,7 @@
 #include <nebula/client/ConnectionPool.h>
 #include <nebula/client/Session.h>
 
-#include "../../ClientTest.h"
+#include "./ClientTest.h"
 
 // Require a nebula server could access
 

diff --git a/src/client/tests/ConnectionTest.cpp b/src/client/tests/ConnectionTest.cpp
@@ -10,7 +10,7 @@
 #include <gtest/gtest.h>
 #include <nebula/client/Connection.h>
 
-#include "../../ClientTest.h"
+#include "./ClientTest.h"
 
 // Require a nebula server could access
 

diff --git a/src/client/tests/SessionTest.cpp b/src/client/tests/SessionTest.cpp
@@ -18,7 +18,7 @@
 #include <chrono>
 #include <thread>
 
-#include "../../ClientTest.h"
+#include "./ClientTest.h"
 
 // Require a nebula server could access
 

diff --git a/src/client/tests/TimezoneConversionTest.cpp b/src/client/tests/TimezoneConversionTest.cpp
@@ -15,7 +15,7 @@
 #include <chrono>
 #include <thread>
 
-#include "../../ClientTest.h"
+#include "./ClientTest.h"
 
 // Require a nebula server could access
 

diff --git a/src/mclient/MetaClient.cpp b/src/mclient/MetaClient.cpp
@@ -15,18 +15,19 @@
 
 namespace nebula {
 
-MetaClient::MetaClient(const std::vector<std::string>& metaAddrs) {
+MetaClient::MetaClient(const std::vector<std::string>& metaAddrs, const MConfig& mConfig) {
   for (const auto& addr : metaAddrs) {
     std::vector<std::string> ip_port;
     folly::split(':', addr, ip_port, true);
     CHECK(ip_port.size() == 2) << "meta server addr " << addr << " is illegal";
     metaAddrs_.emplace_back(ip_port[0], folly::to<int32_t>(ip_port[1]));
   }
   CHECK(!metaAddrs_.empty()) << "metaAddrs_ is empty";
+  mConfig_ = mConfig;
 
   ioExecutor_ = std::make_shared<folly::IOThreadPoolExecutor>(std::thread::hardware_concurrency());
-  clientsMan_ =
-      std::make_shared<thrift::ThriftClientManager<meta::cpp2::MetaServiceAsyncClient>>(false);
+  clientsMan_ = std::make_shared<thrift::ThriftClientManager<meta::cpp2::MetaServiceAsyncClient>>(
+      mConfig_.connTimeoutInMs_, mConfig_.enableSSL_, mConfig_.CAPath_);
   bool b = loadData();  // load data into cache
   if (!b) {
     LOG(ERROR) << "load data failed";
@@ -199,8 +200,7 @@ void MetaClient::getResponse(Request req,
               respGen = std::move(respGen),
               pro = std::move(pro),
               this]() mutable {
-               auto client = clientsMan_->client(
-                   host, evb, false, 60 * 1000);  // FLAGS_meta_client_timeout_ms
+               auto client = clientsMan_->client(host, evb, false, mConfig_.clientTimeoutInMs_);
                LOG(INFO) << "Send request to meta " << host;
                remoteFunc(client, req)
                    .via(evb)

diff --git a/src/mclient/tests/MetaClientTest.cpp b/src/mclient/tests/MetaClientTest.cpp
@@ -11,17 +11,16 @@
 #include <nebula/client/ConnectionPool.h>
 #include <nebula/client/Session.h>
 #include <nebula/mclient/MetaClient.h>
-#include <nebula/sclient/Init.h>
 
-#include "../../ClientTest.h"
+#include "./MClientTest.h"
 #include "common/datatypes/HostAddr.h"
 #include "common/thrift/ThriftTypes.h"
 
 // Require a nebula server could access
 
 #define kServerHost "127.0.0.1"
 
-class MetaClientTest : public ClientTest {
+class MetaClientTest : public MClientTest {
  protected:
   static void prepare() {
     nebula::ConnectionPool pool;

diff --git a/src/sclient/StorageClient.cpp b/src/sclient/StorageClient.cpp
@@ -13,12 +13,15 @@
 
 namespace nebula {
 
-StorageClient::StorageClient(const std::vector<std::string>& metaAddrs) {
-  mClient_ = std::make_unique<MetaClient>(metaAddrs);
+StorageClient::StorageClient(const std::vector<std::string>& metaAddrs,
+                             const MConfig& mConfig,
+                             const SConfig& sConfig) {
+  mClient_ = std::make_unique<MetaClient>(metaAddrs, mConfig);
+  sConfig_ = sConfig;
   ioExecutor_ = std::make_shared<folly::IOThreadPoolExecutor>(std::thread::hardware_concurrency());
   clientsMan_ =
       std::make_shared<thrift::ThriftClientManager<storage::cpp2::GraphStorageServiceAsyncClient>>(
-          false);
+          sConfig.connTimeoutInMs_, sConfig.enableSSL_, sConfig.CAPath_);
 }
 
 StorageClient::~StorageClient() = default;
@@ -118,8 +121,7 @@ void StorageClient::getResponse(std::pair<HostAddr, Request>&& request,
               pro = std::move(pro),
               this]() mutable {
                auto host = request.first;
-               auto client = clientsMan_->client(
-                   host, evb, false, 60 * 1000);  // FLAGS_storage_client_timeout_ms
+               auto client = clientsMan_->client(host, evb, false, sConfig_.clientTimeoutInMs_);
                LOG(INFO) << "Send request to storage " << host;
                remoteFunc(client.get(), request.second)
                    .via(evb)

diff --git a/src/sclient/tests/StorageClientTest.cpp b/src/sclient/tests/StorageClientTest.cpp
@@ -3,6 +3,7 @@
  * This source code is licensed under Apache 2.0 License.
  */
 
+#include <common/Init.h>
 #include <common/datatypes/DataSet.h>
 #include <folly/synchronization/Baton.h>
 #include <glog/logging.h>
@@ -11,17 +12,16 @@
 #include <nebula/client/ConnectionPool.h>
 #include <nebula/client/Session.h>
 #include <nebula/mclient/MetaClient.h>
-#include <nebula/sclient/Init.h>
 #include <nebula/sclient/ScanEdgeIter.h>
 #include <nebula/sclient/StorageClient.h>
 
-#include "../../ClientTest.h"
+#include "./SClientTest.h"
 
 // Require a nebula server could access
 
 #define kServerHost "127.0.0.1"
 
-class StorageClientTest : public ClientTest {
+class StorageClientTest : public SClientTest {
  protected:
   static void prepare() {
     nebula::ConnectionPool pool;

diff --git a/src/thrift/ThriftClientManager-inl.h b/src/thrift/ThriftClientManager-inl.h
@@ -10,6 +10,8 @@
 #include <folly/system/ThreadName.h>
 #include <thrift/lib/cpp2/async/HeaderClientChannel.h>
 
+#include "../SSLConfig.h"
+
 namespace nebula {
 namespace thrift {
 
@@ -67,14 +69,13 @@ std::shared_ptr<ClientType> ThriftClientManager<ClientType>::client(const HostAd
 
   VLOG(2) << "Connecting to " << host << " for " << ++connectionCount << " times";
   std::shared_ptr<folly::AsyncSocket> socket;
-  evb->runImmediatelyOrRunInEventBaseThreadAndWait([&socket, evb, resolved]() {
-    // if (enableSSL_) {
-    //   socket = folly::AsyncSSLSocket::newSocket(nebula::createSSLContext(), evb);
-    //   socket->connect(nullptr, resolved.host, resolved.port, 1000);  // FLAGS_conn_timeout_ms
-    // } else {
-    socket = folly::AsyncSocket::newSocket(
-        evb, resolved.host, resolved.port, 1000);  // FLAGS_conn_timeout_ms
-    // }
+  evb->runImmediatelyOrRunInEventBaseThreadAndWait([this, &socket, evb, resolved]() {
+    if (enableSSL_) {
+      socket = folly::AsyncSSLSocket::newSocket(nebula::createSSLContext(CAPath_), evb);
+      socket->connect(nullptr, resolved.host, resolved.port, connTimeoutInMs_);
+    } else {
+      socket = folly::AsyncSocket::newSocket(evb, resolved.host, resolved.port, connTimeoutInMs_);
+    }
   });
   auto headerClientChannel = apache::thrift::HeaderClientChannel::newChannel(socket);
   if (timeout > 0) {