rebase with latest code

Author: bajajneha27

google/cloud/internal/rest_pure_background_threads_impl.h

@@ -52,12 +52,6 @@ struct RestPureQueueTraits {
   }
 };
 
-/// Background threads that run on a RestPureCompletionQueue.
-using AutomaticallyCreatedRestPureBackgroundThreads =
-    google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<
-        RestPureCompletionQueue, RestPureBackgroundThreads,
-        RestPureQueueTraits>;
-
 /// Background threads that run on a RestPureCompletionQueue.
 using AutomaticallyCreatedRestPureBackgroundThreads =
     google::cloud::internal::AutomaticallyCreatedBackgroundThreadsImpl<

google/cloud/storage/grpc_plugin.cc

@@ -38,10 +38,8 @@ google::cloud::storage::Client MakeGrpcClient(Options opts) {
   auto stub = std::make_unique<storage_internal::GrpcStub>(opts);
   storage_internal::TracingConnection::AsyncRunner runner;
   if (google::cloud::internal::TracingEnabled(opts)) {
-    auto threads = std::shared_ptr<google::cloud::BackgroundThreads>(
-        google::cloud::internal::MakeBackgroundThreadsFactory(opts)());
-    runner = [threads](std::function<void()> f) {
-      threads->cq().RunAsync(std::move(f));
+    runner = [cq = stub->cq()](std::function<void()> f) mutable {
+      cq.RunAsync(std::move(f));
     };
   }
   return storage::internal::ClientImplDetails::CreateWithoutDecorations(

google/cloud/storage/internal/grpc/stub.h

@@ -51,6 +51,7 @@ class GrpcStub : public GenericStub {
       google::storage::v2::WriteObjectResponse>;
 
   Options options() const override;
+  google::cloud::CompletionQueue cq() const { return background_->cq(); }
 
   StatusOr<storage::internal::ListBucketsResponse> ListBuckets(
       rest_internal::RestContext& context, Options const& options,

google/cloud/storage/internal/tracing_connection.cc

@@ -30,19 +30,25 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
 TracingConnection::TracingConnection(std::shared_ptr<StorageConnection> impl,
                                      AsyncRunner runner)
-    : impl_(std::move(impl)), runner_(std::move(runner)) {
-  if (!runner_) {
-    auto threads =
-        std::make_shared<google::cloud::rest_internal::
-                             AutomaticallyCreatedRestPureBackgroundThreads>(1U);
-    runner_ = [threads](std::function<void()> f) {
-      threads->cq().RunAsync(std::move(f));
-    };
-  }
-}
+    : impl_(std::move(impl)), runner_(std::move(runner)) {}
 
 TracingConnection::~TracingConnection() = default;
 
+TracingConnection::AsyncRunner const& TracingConnection::runner() {
+  absl::call_once(once_flag_, [this] {
+    if (!runner_) {
+      auto threads =
+          std::make_shared<google::cloud::rest_internal::
+                               AutomaticallyCreatedRestPureBackgroundThreads>(
+              1U);
+      runner_ = [threads](std::function<void()> f) {
+        threads->cq().RunAsync(std::move(f));
+      };
+    }
+  });
+  return runner_;
+}
+
 BucketMetadataCache& TracingConnection::cache() {
   static BucketMetadataCache instance(10000);
   return instance;
@@ -65,7 +71,7 @@ void TracingConnection::MaybeTriggerBackgroundFetch(
   }
 
   auto current_options = google::cloud::internal::SaveCurrentOptions();
-  runner_([this, bucket_name, current_options]() {
+  runner()([this, bucket_name, current_options]() {
     google::cloud::internal::OptionsSpan span(current_options);
     storage::internal::GetBucketMetadataRequest request(bucket_name);
     auto result = impl_->GetBucketMetadata(request);

google/cloud/storage/internal/tracing_connection.h

@@ -19,6 +19,7 @@
 #include "google/cloud/storage/internal/storage_connection.h"
 #include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/storage/version.h"
+#include "absl/base/call_once.h"
 #include <functional>
 #include <memory>
 #include <string>
@@ -206,8 +207,11 @@ class TracingConnection : public storage::internal::StorageConnection {
 
   static BucketMetadataCache& cache();
 
+  AsyncRunner const& runner();
+
   std::shared_ptr<StorageConnection> impl_;
   AsyncRunner runner_;
+  absl::once_flag once_flag_;
 };
 
 std::shared_ptr<storage::internal::StorageConnection> MakeTracingClient(

google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc

@@ -54,8 +54,8 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
   // The regional_bucket was created in the us-west2 region.
   auto options = Options{}.set<RestEndpointOption>(
       "https://storage.us-west2.rep.googleapis.com");
-  auto client = MakeIntegrationTestClient(options);
 
+  auto client = MakeIntegrationTestClient(options);
   auto object_name = MakeRandomObjectName();
   std::string expected = LoremIpsum();
 

google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc

@@ -47,6 +47,7 @@ TEST_F(ObjectPlentyClientsSimultaneouslyIntegrationTest,
 
   auto client = MakeIntegrationTestClient();
   auto object_name = MakeRandomObjectName();
+
   std::string expected = LoremIpsum();
 
   // Create the object, but only if it does not exist already.