C++ migration: make C++ Datastore create gRPC objects (#1765)

Also introduce a new class, `GrpcConnection`, somewhat similar to the class of the same name in Web client, which does the bulk of the work making sure there is a valid gRPC channel/stub.

Author: var-const

Firestore/Example/Firestore.xcodeproj/project.pbxproj

@@ -37,7 +37,7 @@
 		54511E8E209805F8005BD28F /* hashing_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = 54511E8D209805F8005BD28F /* hashing_test.cc */; };
 		5467FB01203E5717009C9584 /* FIRFirestoreTests.mm in Sources */ = {isa = PBXBuildFile; fileRef = 5467FAFF203E56F8009C9584 /* FIRFirestoreTests.mm */; };
 		5467FB08203E6A44009C9584 /* app_testing.mm in Sources */ = {isa = PBXBuildFile; fileRef = 5467FB07203E6A44009C9584 /* app_testing.mm */; };
-		546854AA20A36867004BDBD5 /* datastore_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = 546854A820A36867004BDBD5 /* datastore_test.cc */; };
+		546854AA20A36867004BDBD5 /* datastore_test.mm in Sources */ = {isa = PBXBuildFile; fileRef = 546854A820A36867004BDBD5 /* datastore_test.mm */; };
 		54740A571FC914BA00713A1A /* secure_random_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = 54740A531FC913E500713A1A /* secure_random_test.cc */; };
 		54740A581FC914F000713A1A /* autoid_test.cc in Sources */ = {isa = PBXBuildFile; fileRef = 54740A521FC913E500713A1A /* autoid_test.cc */; };
 		54764FAF1FAA21B90085E60A /* FSTGoogleTestTests.mm in Sources */ = {isa = PBXBuildFile; fileRef = 54764FAE1FAA21B90085E60A /* FSTGoogleTestTests.mm */; };
@@ -298,8 +298,8 @@
 		132E3BB3D5C42282B4ACFB20 /* FSTLevelDBBenchmarkTests.mm */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.objcpp; path = FSTLevelDBBenchmarkTests.mm; sourceTree = "<group>"; };
 		2A0CF41BA5AED6049B0BEB2C /* type_traits_apple_test.mm */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.cpp.objcpp; path = type_traits_apple_test.mm; sourceTree = "<group>"; };
 		2B50B3A0DF77100EEE887891 /* Pods_Firestore_Tests_iOS.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = Pods_Firestore_Tests_iOS.framework; sourceTree = BUILT_PRODUCTS_DIR; };
-		332485C4DCC6BA0DBB5E31B7 /* leveldb_util_test.cc */ = {isa = PBXFileReference; includeInIndex = 1; path = leveldb_util_test.cc; sourceTree = "<group>"; };
-		3564F6872205C4F7001D5436 /* grpc_stream_tester.cc */ = {isa = PBXFileReference; includeInIndex = 1; path = grpc_stream_tester.cc; sourceTree = "<group>"; };
+		332485C4DCC6BA0DBB5E31B7 /* leveldb_util_test.cc */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.cpp.cpp; path = leveldb_util_test.cc; sourceTree = "<group>"; };
+		3564F6872205C4F7001D5436 /* grpc_stream_tester.cc */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.cpp.cpp; path = grpc_stream_tester.cc; sourceTree = "<group>"; };
 		358C3B5FE573B1D60A4F7592 /* strerror_test.cc */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.cpp.cpp; path = strerror_test.cc; sourceTree = "<group>"; };
 		3B843E4A1F3930A400548890 /* remote_store_spec_test.json */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text.json; path = remote_store_spec_test.json; sourceTree = "<group>"; };
 		3C81DE3772628FE297055662 /* Pods-Firestore_Example_iOS.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-Firestore_Example_iOS.debug.xcconfig"; path = "Pods/Target Support Files/Pods-Firestore_Example_iOS/Pods-Firestore_Example_iOS.debug.xcconfig"; sourceTree = "<group>"; };
@@ -311,7 +311,7 @@
 		5467FAFF203E56F8009C9584 /* FIRFirestoreTests.mm */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.objcpp; path = FIRFirestoreTests.mm; sourceTree = "<group>"; };
 		5467FB06203E6A44009C9584 /* app_testing.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = app_testing.h; sourceTree = "<group>"; };
 		5467FB07203E6A44009C9584 /* app_testing.mm */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.objcpp; path = app_testing.mm; sourceTree = "<group>"; };
-		546854A820A36867004BDBD5 /* datastore_test.cc */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = datastore_test.cc; sourceTree = "<group>"; };
+		546854A820A36867004BDBD5 /* datastore_test.mm */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.objcpp; path = datastore_test.mm; sourceTree = "<group>"; };
 		54740A521FC913E500713A1A /* autoid_test.cc */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = autoid_test.cc; sourceTree = "<group>"; };
 		54740A531FC913E500713A1A /* secure_random_test.cc */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = secure_random_test.cc; sourceTree = "<group>"; };
 		54764FAE1FAA21B90085E60A /* FSTGoogleTestTests.mm */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.objcpp; path = FSTGoogleTestTests.mm; sourceTree = "<group>"; };
@@ -646,7 +646,7 @@
 		546854A720A3681B004BDBD5 /* remote */ = {
 			isa = PBXGroup;
 			children = (
-				546854A820A36867004BDBD5 /* datastore_test.cc */,
+				546854A820A36867004BDBD5 /* datastore_test.mm */,
 				B6D1B68420E2AB1A00B35856 /* exponential_backoff_test.cc */,
 				B6BBE42F21262CF400C6A53E /* grpc_stream_test.cc */,
 				61F72C5520BC48FD001A68CB /* serializer_test.cc */,
@@ -1877,13 +1877,12 @@
 				B6FB467D208E9D3C00554BA2 /* async_queue_test.cc in Sources */,
 				54740A581FC914F000713A1A /* autoid_test.cc in Sources */,
 				AB380D02201BC69F00D97691 /* bits_test.cc in Sources */,
-				B646EE6121224220001A1F18 /* buffered_writer_test.cc in Sources */,
 				618BBEA920B89AAC00B5BCE7 /* common.pb.cc in Sources */,
 				548DB929200D59F600E00ABC /* comparison_test.cc in Sources */,
 				ABC1D7DC2023A04B00BA84F0 /* credentials_provider_test.cc in Sources */,
 				ABE6637A201FA81900ED349A /* database_id_test.cc in Sources */,
 				AB38D93020236E21000A432D /* database_info_test.cc in Sources */,
-				546854AA20A36867004BDBD5 /* datastore_test.cc in Sources */,
+				546854AA20A36867004BDBD5 /* datastore_test.mm in Sources */,
 				618BBEAC20B89AAC00B5BCE7 /* document.pb.cc in Sources */,
 				B6152AD7202A53CB000E5744 /* document_key_test.cc in Sources */,
 				AB6B908420322E4D00CC290A /* document_test.cc in Sources */,

Firestore/core/src/firebase/firestore/remote/datastore.h

@@ -17,7 +17,18 @@
 #ifndef FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_DATASTORE_H_
 #define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_DATASTORE_H_
 
+#include <memory>
+#include <string>
+
+#include "Firestore/core/src/firebase/firestore/core/database_info.h"
+#include "Firestore/core/src/firebase/firestore/remote/grpc_connection.h"
+#include "Firestore/core/src/firebase/firestore/remote/grpc_stream.h"
+#include "Firestore/core/src/firebase/firestore/remote/grpc_stream_observer.h"
+#include "Firestore/core/src/firebase/firestore/util/async_queue.h"
+#include "Firestore/core/src/firebase/firestore/util/executor.h"
 #include "Firestore/core/src/firebase/firestore/util/status.h"
+#include "absl/strings/string_view.h"
+#include "grpcpp/completion_queue.h"
 #include "grpcpp/support/status.h"
 
 namespace firebase {
@@ -26,12 +37,36 @@ namespace remote {
 
 class Datastore {
  public:
-  static util::Status ConvertStatus(grpc::Status grpc_error);
+  Datastore(const core::DatabaseInfo& database_info,
+            util::AsyncQueue* worker_queue);
+
+  void Shutdown();
+
+  std::unique_ptr<GrpcStream> CreateGrpcStream(absl::string_view rpc_name,
+                                               absl::string_view token,
+                                               GrpcStreamObserver* observer);
+
+  static util::Status ConvertStatus(grpc::Status from);
+
+  static std::string GetWhitelistedHeadersAsString(
+      const GrpcStream::MetadataT& headers);
 
   Datastore(const Datastore& other) = delete;
   Datastore(Datastore&& other) = delete;
   Datastore& operator=(const Datastore& other) = delete;
   Datastore& operator=(Datastore&& other) = delete;
+
+ private:
+  void PollGrpcQueue();
+
+  static GrpcStream::MetadataT ExtractWhitelistedHeaders(
+      const GrpcStream::MetadataT& headers);
+
+  // A separate executor dedicated to polling gRPC completion queue (which is
+  // shared for all spawned `GrpcStream`s).
+  std::unique_ptr<util::internal::Executor> rpc_executor_;
+  grpc::CompletionQueue grpc_queue_;
+  GrpcConnection grpc_connection_;
 };
 
 }  // namespace remote

Firestore/core/src/firebase/firestore/remote/datastore.mm

@@ -16,16 +16,85 @@
 
 #include "Firestore/core/src/firebase/firestore/remote/datastore.h"
 
+#include <unordered_set>
+
 #include "Firestore/core/include/firebase/firestore/firestore_errors.h"
+#include "Firestore/core/src/firebase/firestore/core/database_info.h"
+#include "Firestore/core/src/firebase/firestore/remote/grpc_completion.h"
+#include "Firestore/core/src/firebase/firestore/util/executor_libdispatch.h"
 #include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
+#include "absl/memory/memory.h"
+#include "absl/strings/str_cat.h"
 
 namespace firebase {
 namespace firestore {
 namespace remote {
 
-util::Status Datastore::ConvertStatus(grpc::Status from) {
+using core::DatabaseInfo;
+using util::AsyncQueue;
+using util::Status;
+using util::internal::Executor;
+using util::internal::ExecutorLibdispatch;
+
+namespace {
+
+std::unique_ptr<Executor> CreateExecutor() {
+  auto queue = dispatch_queue_create("com.google.firebase.firestore.rpc",
+                                     DISPATCH_QUEUE_SERIAL);
+  return absl::make_unique<ExecutorLibdispatch>(queue);
+}
+
+std::string MakeString(grpc::string_ref grpc_str) {
+  return {grpc_str.begin(), grpc_str.size()};
+}
+
+absl::string_view MakeStringView(grpc::string_ref grpc_str) {
+  return {grpc_str.begin(), grpc_str.size()};
+}
+
+}  // namespace
+
+Datastore::Datastore(const DatabaseInfo &database_info,
+                     AsyncQueue *worker_queue)
+    : grpc_connection_{database_info, worker_queue, &grpc_queue_},
+      rpc_executor_{CreateExecutor()} {
+  rpc_executor_->Execute([this] { PollGrpcQueue(); });
+}
+
+void Datastore::Shutdown() {
+  // `grpc::CompletionQueue::Next` will only return `false` once `Shutdown` has
+  // been called and all submitted tags have been extracted. Without this call,
+  // `rpc_executor_` will never finish.
+  grpc_queue_.Shutdown();
+  // Drain the executor to make sure it extracted all the operations from gRPC
+  // completion queue.
+  rpc_executor_->ExecuteBlocking([] {});
+}
+
+void Datastore::PollGrpcQueue() {
+  HARD_ASSERT(rpc_executor_->IsCurrentExecutor(),
+              "PollGrpcQueue should only be called on the "
+              "dedicated Datastore executor");
+
+  void *tag = nullptr;
+  bool ok = false;
+  while (grpc_queue_.Next(&tag, &ok)) {
+    auto completion = static_cast<GrpcCompletion *>(tag);
+    HARD_ASSERT(tag, "gRPC queue returned a null tag");
+    completion->Complete(ok);
+  }
+}
+
+std::unique_ptr<GrpcStream> Datastore::CreateGrpcStream(
+    absl::string_view rpc_name,
+    absl::string_view token,
+    GrpcStreamObserver *observer) {
+  return grpc_connection_.CreateStream(token, rpc_name, observer);
+}
+
+Status Datastore::ConvertStatus(grpc::Status from) {
   if (from.ok()) {
-    return {};
+    return Status::OK();
   }
 
   grpc::StatusCode error_code = from.error_code();
@@ -36,6 +105,22 @@
   return {static_cast<FirestoreErrorCode>(error_code), from.error_message()};
 }
 
+std::string Datastore::GetWhitelistedHeadersAsString(
+    const GrpcStream::MetadataT &headers) {
+  static std::unordered_set<std::string> whitelist = {
+      "date", "x-google-backends", "x-google-netmon-label", "x-google-service",
+      "x-google-gfe-request-trace"};
+
+  std::string result;
+  for (const auto &kv : headers) {
+    if (whitelist.find(MakeString(kv.first)) != whitelist.end()) {
+      absl::StrAppend(&result, MakeStringView(kv.first), ": ",
+                      MakeStringView(kv.second), "\n");
+    }
+  }
+  return result;
+}
+
 }  // namespace remote
 }  // namespace firestore
 }  // namespace firebase

Firestore/core/src/firebase/firestore/remote/grpc_connection.h

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2018 Google
+ *
+ * Licensed 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.
+ */
+
+#ifndef FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_GRPC_CONNECTION_H_
+#define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_GRPC_CONNECTION_H_
+
+#include <memory>
+
+#include "Firestore/core/src/firebase/firestore/core/database_info.h"
+#include "Firestore/core/src/firebase/firestore/remote/grpc_stream.h"
+#include "Firestore/core/src/firebase/firestore/remote/grpc_stream_observer.h"
+#include "absl/strings/string_view.h"
+#include "grpcpp/channel.h"
+#include "grpcpp/client_context.h"
+#include "grpcpp/completion_queue.h"
+#include "grpcpp/generic/generic_stub.h"
+
+namespace firebase {
+namespace firestore {
+namespace remote {
+
+// PORTING NOTE: this class has limited resemblance to `GrpcConnection` in Web
+// client. However, unlike Web client, it's not meant to hide different
+// implementations of a `Connection` under a single interface.
+
+/**
+ * Creates and owns gRPC objects (channel and stub) necessary to produce a
+ * `GrpcStream`.
+ */
+class GrpcConnection {
+ public:
+  GrpcConnection(const core::DatabaseInfo& database_info,
+                 util::AsyncQueue* worker_queue,
+                 grpc::CompletionQueue* grpc_queue);
+
+  /**
+   * Creates a stream to the given stream RPC endpoint. The resulting stream
+   * needs to be `Start`ed before it can be used.
+   */
+  // PORTING NOTE: unlike Web client, the created stream is not open and has to
+  // be started manually.
+  std::unique_ptr<GrpcStream> CreateStream(absl::string_view rpc_name,
+                                           absl::string_view token,
+                                           GrpcStreamObserver* observer);
+
+ private:
+  std::unique_ptr<grpc::ClientContext> CreateContext(
+      absl::string_view token) const;
+  void EnsureActiveStub();
+
+  const core::DatabaseInfo* database_info_ = nullptr;
+  util::AsyncQueue* worker_queue_ = nullptr;
+  grpc::CompletionQueue* grpc_queue_ = nullptr;
+
+  std::shared_ptr<grpc::Channel> grpc_channel_;
+  std::unique_ptr<grpc::GenericStub> grpc_stub_;
+};
+
+}  // namespace remote
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_GRPC_CONNECTION_H_