impl(v3): add generator support for connection cq access

Author: scotthart

generator/generator_config.proto

@@ -167,6 +167,12 @@ message ServiceConfiguration {
   // RPCs. If set to false (the default), a no-op resumption function will be
   // generated.
   bool omit_streaming_updater = 29;
+
+  // In rare cases, specifically bigtable::WaitForConsistency, the
+  // CompletionQueue from the BackgroundThreads owned by the Connection is
+  // needed elsewhere. This emits a protected accessor and friend function for
+  // that purpose.
+  bool emit_completion_queue_accessor = 30;
 }
 
 message DiscoveryDocumentDefinedProduct {

generator/generator_config.textproto

@@ -588,6 +588,7 @@ service {
     {rpc_name: "BigtableTableAdmin.CheckConsistency", idempotency: IDEMPOTENT}
   ]
   omit_repo_metadata: true
+  emit_completion_queue_accessor: true
 }
 
 # Billing

generator/internal/connection_generator.cc

@@ -57,6 +57,8 @@ Status ConnectionGenerator::GenerateHeader() {
       {vars("idempotency_policy_header_path"), vars("retry_traits_header_path"),
        HasLongrunningMethod() ? "google/cloud/no_await_tag.h" : "",
        IsExperimental() ? "google/cloud/experimental_tag.h" : "",
+       HasEmitCompletionQueueAccessor() ? "google/cloud/completion_queue.h"
+                                        : "",
        "google/cloud/backoff_policy.h",
        HasLongrunningMethod() || HasAsyncMethod() ? "google/cloud/future.h"
                                                   : "",
@@ -91,7 +93,21 @@ Status ConnectionGenerator::GenerateHeader() {
     }
   }
 
-  auto result = HeaderOpenNamespaces();
+  Status result;
+  if (HasEmitCompletionQueueAccessor()) {
+    result = HeaderOpenNamespaces();
+    if (!result.ok()) return result;
+    HeaderPrint(R"""(class $connection_class_name$;)""");
+    HeaderCloseNamespaces();
+
+    result = HeaderOpenNamespaces(NamespaceType::kInternal);
+    if (!result.ok()) return result;
+    HeaderPrint(
+        R"""(StatusOr<CompletionQueue> completion_queue($product_namespace$::$connection_class_name$ const& conn);)""");
+    HeaderCloseNamespaces();
+  }
+
+  result = HeaderOpenNamespaces();
   if (!result.ok()) return result;
 
   HeaderPrint(R"""(
@@ -315,6 +331,13 @@ class $connection_class_name$ {
         __FILE__, __LINE__);
   }
 
+  if (HasEmitCompletionQueueAccessor()) {
+    HeaderPrint(R"""( protected:
+  friend StatusOr<CompletionQueue> $product_internal_namespace$::completion_queue(
+      $connection_class_name$ const& conn);
+  virtual StatusOr<CompletionQueue> completion_queue() const;
+)""");
+  }
   // close abstract interface Connection base class
   HeaderPrint("};\n");
 
@@ -490,6 +513,15 @@ future<StatusOr<$response_type$>>
         __FILE__, __LINE__);
   }
 
+  if (HasEmitCompletionQueueAccessor()) {
+    CcPrint(
+        R"""(
+StatusOr<CompletionQueue> $connection_class_name$::completion_queue() const {
+  return Status(StatusCode::kUnimplemented, "not implemented");
+}
+)""");
+  }
+
   if (HasGenerateGrpcTransport()) {
     EmitFactoryFunctionDefinition(EndpointLocationStyle());
   }

generator/internal/connection_impl_generator.cc

@@ -133,6 +133,13 @@ class $connection_class_name$Impl
           std::make_shared<google::cloud::internal::InvocationIdGenerator>();)""");
   }
 
+  if (HasEmitCompletionQueueAccessor()) {
+    HeaderPrint(
+        R"""(
+  StatusOr<CompletionQueue> completion_queue() const override;
+)""");
+  }
+
   // This closes the *ConnectionImpl class definition.
   HeaderPrint("\n};\n");
 
@@ -208,6 +215,15 @@ std::unique_ptr<PollingPolicy> polling_policy(Options const& options) {
 } // namespace
 )""");
 
+  if (HasEmitCompletionQueueAccessor()) {
+    CcPrint(R"""(
+StatusOr<CompletionQueue> completion_queue(
+    $product_namespace$::$connection_class_name$ const& conn) {
+  return conn.completion_queue();
+}
+)""");
+  }
+
   // streaming updater functions
   if (!OmitStreamingUpdater(vars())) {
     for (auto const& method : methods()) {
@@ -242,6 +258,14 @@ void $service_name$$method_name$StreamingUpdater(
     CcPrintMethod(method, __FILE__, __LINE__, AsyncMethodDefinition(method));
   }
 
+  if (HasEmitCompletionQueueAccessor()) {
+    CcPrint(R"""(
+StatusOr<CompletionQueue> $connection_class_name$Impl::completion_queue() const {
+  return background_->cq();
+}
+)""");
+  }
+
   CcCloseNamespaces();
   return {};
 }

generator/internal/service_code_generator.cc

@@ -511,6 +511,11 @@ bool ServiceCodeGenerator::IsDeprecated() const {
   return service_descriptor_->options().deprecated();
 }
 
+bool ServiceCodeGenerator::HasEmitCompletionQueueAccessor() const {
+  return vars().find("emit_completion_queue_accessor") != vars().end() &&
+         vars().at("emit_completion_queue_accessor") == "true";
+}
+
 }  // namespace generator_internal
 }  // namespace cloud
 }  // namespace google

generator/internal/service_code_generator.h

@@ -246,6 +246,8 @@ class ServiceCodeGenerator : public GeneratorInterface {
    */
   bool IsDeprecated() const;
 
+  bool HasEmitCompletionQueueAccessor() const;
+
  private:
   void SetMethods();
 

generator/standalone_main.cc

@@ -296,6 +296,11 @@ std::vector<std::future<google::cloud::Status>> GenerateCodeFromProtos(
     if (service.omit_streaming_updater()) {
       args.emplace_back("--cpp_codegen_opt=omit_streaming_updater=true");
     }
+    if (service.emit_completion_queue_accessor()) {
+      args.emplace_back(
+          "--cpp_codegen_opt=emit_completion_queue_accessor=true");
+    }
+
     if (service.generate_round_robin_decorator()) {
       args.emplace_back(
           "--cpp_codegen_opt=generate_round_robin_decorator=true");