feat(pubsub): support x-goog-user-project (#8456)

Change the metadata decorators to inject `x-goog-user-project` from the
current option span.

For the admin APIs, use the per-call options to configure a span in the
`*Client` class.  For `pubsub::Publisher` and `pubsub::Subscriber` it is
less clear which call "triggers" an RPC, and far less clear how to merge
options from multiple RPCs.  I have decided to use the per-connection
options in this case. We can always add more fine-grained options later,
but this takes care of the common case.

Author: coryan

google/cloud/pubsub/internal/default_batch_sink.cc

@@ -22,17 +22,17 @@ namespace pubsub_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
 DefaultBatchSink::DefaultBatchSink(std::shared_ptr<PublisherStub> stub,
-                                   CompletionQueue cq, Options const& opts)
-    : stub_(std::move(stub)),
-      cq_(std::move(cq)),
-      retry_policy_(opts.get<pubsub::RetryPolicyOption>()->clone()),
-      backoff_policy_(opts.get<pubsub::BackoffPolicyOption>()->clone()) {}
+                                   CompletionQueue cq, Options opts)
+    : stub_(std::move(stub)), cq_(std::move(cq)), options_(std::move(opts)) {}
 
 future<StatusOr<google::pubsub::v1::PublishResponse>>
 DefaultBatchSink::AsyncPublish(google::pubsub::v1::PublishRequest request) {
+  internal::OptionsSpan span(options_);
+
   auto& stub = stub_;
   return internal::AsyncRetryLoop(
-      retry_policy_->clone(), backoff_policy_->clone(),
+      options_.get<pubsub::RetryPolicyOption>()->clone(),
+      options_.get<pubsub::BackoffPolicyOption>()->clone(),
       Idempotency::kIdempotent, cq_,
       [stub](CompletionQueue& cq, std::unique_ptr<grpc::ClientContext> context,
              google::pubsub::v1::PublishRequest const& request) {

google/cloud/pubsub/internal/default_batch_sink.h

@@ -32,10 +32,9 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 class DefaultBatchSink : public BatchSink {
  public:
   static std::shared_ptr<DefaultBatchSink> Create(
-      std::shared_ptr<PublisherStub> stub, CompletionQueue cq,
-      Options const& opts) {
+      std::shared_ptr<PublisherStub> stub, CompletionQueue cq, Options opts) {
     return std::shared_ptr<DefaultBatchSink>(
-        new DefaultBatchSink(std::move(stub), std::move(cq), opts));
+        new DefaultBatchSink(std::move(stub), std::move(cq), std::move(opts)));
   }
 
   ~DefaultBatchSink() override = default;
@@ -47,12 +46,11 @@ class DefaultBatchSink : public BatchSink {
 
  private:
   DefaultBatchSink(std::shared_ptr<PublisherStub> stub, CompletionQueue cq,
-                   Options const& opts);
+                   Options opts);
 
   std::shared_ptr<PublisherStub> stub_;
   CompletionQueue cq_;
-  std::unique_ptr<pubsub::RetryPolicy const> retry_policy_;
-  std::unique_ptr<pubsub::BackoffPolicy const> backoff_policy_;
+  Options options_;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/pubsub/internal/publisher_metadata.cc

@@ -95,6 +95,11 @@ void PublisherMetadata::SetMetadata(grpc::ClientContext& context,
                                     std::string const& request_params) {
   context.AddMetadata("x-goog-request-params", request_params);
   context.AddMetadata("x-goog-api-client", x_goog_api_client_);
+  auto const& options = internal::CurrentOptions();
+  if (options.has<UserProjectOption>()) {
+    context.AddMetadata("x-goog-user-project",
+                        options.get<UserProjectOption>());
+  }
 }
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END