feat(generator): merge connection options into client options

Store the Connection options so that they may be made available to
merge into the Client options, and then use them to implement the
four polling-policy data elements.  (This is the generator version
of googleapis#8090.)

Rules of thumb for `Options` values:
- `MergeOptions()` LHS should not include the service defaults
- `MergeOptions()` RHS should include service defaults
- An `OptionsSpan` should include the service defaults
- stored `Options` (data members) should include the service defaults

Code can then normally just look at the `CurrentOptions()`, knowing
that they will include the service defaults.  For example, the policy
options will be available.  A special exception to this, however, is
when testing the connection layer in isolation (without a client), as
no one has created an `OptionsSpan` in that case.  We could add an
`OptionsSpan` to the connection-layer calls, but it would be a no-op
in practice.  So, tests need to ensure required options are available.

Fixes googleapis#8054.

generator/internal/client_generator.cc

@@ -334,7 +334,8 @@ Status ClientGenerator::GenerateCc() {
     "\n"
     "$client_class_name$::$client_class_name$(std::shared_ptr<$connection_class_name$> connection, Options options)"
     " : connection_(std::move(connection)),"
-    " options_($product_internal_namespace$::$service_name$DefaultOptions(std::move(options))) {}\n");
+    " options_(internal::MergeOptions(std::move(options),"
+    " $product_internal_namespace$::$service_name$DefaultOptions(connection_->options()))) {}\n");
   // clang-format on
 
   CcPrint(  // clang-format off

generator/internal/connection_generator.cc

@@ -105,6 +105,10 @@ Status ConnectionGenerator::GenerateHeader() {
     "  virtual ~$connection_class_name$() = 0;\n");
   // clang-format on
 
+  HeaderPrint(R"""(
+  virtual Options options() { return Options{}; }
+)""");
+
   for (auto const& method : methods()) {
     if (IsBidirStreaming(method)) {
       HeaderPrintMethod(method, __FILE__, __LINE__,
@@ -357,7 +361,7 @@ std::shared_ptr<$connection_class_name$> Make$connection_class_name$(
   auto stub = $product_internal_namespace$::CreateDefault$stub_class_name$(
     background->cq(), options);
   return std::make_shared<$product_internal_namespace$::$connection_class_name$Impl>(
-      std::move(background), std::move(stub), options);
+      std::move(background), std::move(stub), std::move(options));
 }
 )""");
 
@@ -371,9 +375,9 @@ std::shared_ptr<$product_namespace$::$connection_class_name$>
 Make$connection_class_name$(
     std::shared_ptr<$stub_class_name$> stub, Options options) {
   options = $service_name$DefaultOptions(std::move(options));
+  auto background = internal::MakeBackgroundThreadsFactory(options)();
   return std::make_shared<$product_internal_namespace$::$connection_class_name$Impl>(
-      internal::MakeBackgroundThreadsFactory(options)(),
-      std::move(stub), std::move(options));
+      std::move(background), std::move(stub), std::move(options));
 }
 )""");
 

generator/internal/connection_impl_generator.cc

@@ -78,7 +78,11 @@ class $connection_class_name$Impl
   $connection_class_name$Impl(
     std::unique_ptr<google::cloud::BackgroundThreads> background,
     std::shared_ptr<$product_internal_namespace$::$stub_class_name$> stub,
-    Options const& options);
+    Options options);
+)""");
+
+  HeaderPrint(R"""(
+  Options options() override { return options_; }
 )""");
 
   for (auto const& method : methods()) {
@@ -90,53 +94,56 @@ class $connection_class_name$Impl
                       AsyncMethodDeclaration(method));
   }
 
+  // `CurrentOptions()` may not have the service default options because we
+  // could be running in a test that calls the ConnectionImpl layer directly,
+  // and it does not create an `internal::OptionsSpan` like the Client layer.
+  // So, we have to fallback to `options_`, which we know has the service
+  // default options because we added them.
   HeaderPrint(R"""(
  private:
   std::unique_ptr<$product_namespace$::$retry_policy_name$> retry_policy() {
     auto const& options = internal::CurrentOptions();
     if (options.has<$product_namespace$::$retry_policy_name$Option>()) {
       return options.get<$product_namespace$::$retry_policy_name$Option>()->clone();
     }
-    return retry_policy_prototype_->clone();
+    return options_.get<$product_namespace$::$retry_policy_name$Option>()->clone();
   }
 
   std::unique_ptr<BackoffPolicy> backoff_policy() {
     auto const& options = internal::CurrentOptions();
     if (options.has<$product_namespace$::$service_name$BackoffPolicyOption>()) {
       return options.get<$product_namespace$::$service_name$BackoffPolicyOption>()->clone();
     }
-    return backoff_policy_prototype_->clone();
+    return options_.get<$product_namespace$::$service_name$BackoffPolicyOption>()->clone();
   }
 
   std::unique_ptr<$product_namespace$::$idempotency_class_name$> idempotency_policy() {
     auto const& options = internal::CurrentOptions();
     if (options.has<$product_namespace$::$idempotency_class_name$Option>()) {
       return options.get<$product_namespace$::$idempotency_class_name$Option>()->clone();
     }
-    return idempotency_policy_->clone();
+    return options_.get<$product_namespace$::$idempotency_class_name$Option>()->
+clone();
   }
-
-  std::unique_ptr<google::cloud::BackgroundThreads> background_;
-  std::shared_ptr<$product_internal_namespace$::$stub_class_name$> stub_;
-  std::unique_ptr<$product_namespace$::$retry_policy_name$ const> retry_policy_prototype_;
-  std::unique_ptr<BackoffPolicy const> backoff_policy_prototype_;
-  std::unique_ptr<$product_namespace$::$idempotency_class_name$> idempotency_policy_;
 )""");
-
   if (HasLongrunningMethod()) {
     HeaderPrint(R"""(
   std::unique_ptr<PollingPolicy> polling_policy() {
     auto const& options = internal::CurrentOptions();
     if (options.has<$product_namespace$::$service_name$PollingPolicyOption>()) {
       return options.get<$product_namespace$::$service_name$PollingPolicyOption>()->clone();
     }
-    return polling_policy_prototype_->clone();
+    return options_.get<$product_namespace$::$service_name$PollingPolicyOption>()->clone();
   }
-
-  std::unique_ptr<PollingPolicy const> polling_policy_prototype_;
 )""");
   }
 
+  HeaderPrint(R"""(
+  std::unique_ptr<google::cloud::BackgroundThreads> background_;
+  std::shared_ptr<$product_internal_namespace$::$stub_class_name$> stub_;
+  Options options_;
+)""");
+
   // This closes the *ConnectionImpl class definition.
   HeaderPrint("};\n");
 
@@ -181,17 +188,9 @@ Status ConnectionImplGenerator::GenerateCc() {
 $connection_class_name$Impl::$connection_class_name$Impl(
     std::unique_ptr<google::cloud::BackgroundThreads> background,
     std::shared_ptr<$product_internal_namespace$::$stub_class_name$> stub,
-    Options const& options)
+    Options options)
   : background_(std::move(background)), stub_(std::move(stub)),
-    retry_policy_prototype_(options.get<$product_namespace$::$retry_policy_name$Option>()->clone()),
-    backoff_policy_prototype_(options.get<$product_namespace$::$service_name$BackoffPolicyOption>()->clone()),
-    idempotency_policy_(options.get<$product_namespace$::$idempotency_class_name$Option>()->clone()))""");
-  // The constructor depends on a couple of things.
-  if (HasLongrunningMethod()) {
-    CcPrint(R"""(,
-    polling_policy_prototype_(options.get<$product_namespace$::$service_name$PollingPolicyOption>()->clone()))""");
-  }
-  CcPrint(R"""( {}
+    options_($product_internal_namespace$::$service_name$DefaultOptions(std::move(options))) {}
 )""");
 
   for (auto const& method : methods()) {

generator/internal/mock_connection_generator.cc

@@ -57,11 +57,9 @@ Status MockConnectionGenerator::GenerateHeader() {
   HeaderPrint(R"""(
 class $mock_connection_class_name$ : public $product_namespace$::$connection_class_name$ {
  public:)""");
-  // Avoid an unsightly blank line at the start of the mock class by not
-  // printing a newline after the `public:` access qualifier. While almost any
-  // class we generate will have some member functions, if there are none, we
-  // need that newline.
-  if (methods().empty() && async_methods().empty()) HeaderPrint("\n");
+  HeaderPrint(R"""(
+  MOCK_METHOD(Options, options, (), (override));
+)""");
 
   for (auto const& method : methods()) {
     if (IsBidirStreaming(method)) {