impl(rest): add completion queue and thread types with no grpc deps

google/cloud/google_cloud_cpp_common.bzl

@@ -74,6 +74,7 @@ google_cloud_cpp_common_hdrs = [
     "internal/retry_info.h",
     "internal/retry_loop_helpers.h",
     "internal/retry_policy_impl.h",
+    "internal/run_async_base.h",
     "internal/service_endpoint.h",
     "internal/sha256_hash.h",
     "internal/sha256_hmac.h",

google/cloud/google_cloud_cpp_common.cmake

@@ -117,6 +117,7 @@ add_library(
     internal/retry_loop_helpers.h
     internal/retry_policy_impl.cc
     internal/retry_policy_impl.h
+    internal/run_async_base.h
     internal/service_endpoint.cc
     internal/service_endpoint.h
     internal/sha256_hash.cc

google/cloud/google_cloud_cpp_rest_internal.bzl

@@ -65,6 +65,8 @@ google_cloud_cpp_rest_internal_hdrs = [
     "internal/rest_opentelemetry.h",
     "internal/rest_options.h",
     "internal/rest_parse_json_error.h",
+    "internal/rest_pure_background_threads_impl.h",
+    "internal/rest_pure_completion_queue_impl.h",
     "internal/rest_request.h",
     "internal/rest_response.h",
     "internal/rest_retry_loop.h",
@@ -121,6 +123,8 @@ google_cloud_cpp_rest_internal_srcs = [
     "internal/rest_lro_helpers.cc",
     "internal/rest_opentelemetry.cc",
     "internal/rest_parse_json_error.cc",
+    "internal/rest_pure_background_threads_impl.cc",
+    "internal/rest_pure_completion_queue_impl.cc",
     "internal/rest_request.cc",
     "internal/rest_response.cc",
     "internal/rest_set_metadata.cc",

google/cloud/google_cloud_cpp_rest_internal.cmake

@@ -113,6 +113,10 @@ add_library(
     internal/rest_options.h
     internal/rest_parse_json_error.cc
     internal/rest_parse_json_error.h
+    internal/rest_pure_background_threads_impl.cc
+    internal/rest_pure_background_threads_impl.h
+    internal/rest_pure_completion_queue_impl.cc
+    internal/rest_pure_completion_queue_impl.h
     internal/rest_request.cc
     internal/rest_request.h
     internal/rest_response.cc

google/cloud/internal/completion_queue_impl.h

@@ -18,6 +18,7 @@
 #include "google/cloud/async_operation.h"
 #include "google/cloud/future.h"
 #include "google/cloud/internal/invoke_result.h"
+#include "google/cloud/internal/run_async_base.h"
 #include "google/cloud/internal/throw_delegate.h"
 #include "google/cloud/status_or.h"
 #include "google/cloud/version.h"
@@ -30,12 +31,6 @@ namespace cloud {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
 
-// Type erase the callables in RunAsync()
-struct RunAsyncBase {
-  virtual ~RunAsyncBase() = default;
-  virtual void exec() = 0;
-};
-
 /**
  * The implementation details for `CompletionQueue`.
  *

google/cloud/internal/rest_completion_queue_impl.cc

@@ -13,50 +13,35 @@
 // limitations under the License.
 
 #include "google/cloud/internal/rest_completion_queue_impl.h"
-#include "google/cloud/internal/timer_queue.h"
 
 namespace google {
 namespace cloud {
 namespace rest_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
 RestCompletionQueueImpl::RestCompletionQueueImpl()
-    : tq_(internal::TimerQueue::Create()) {}
+    : impl_(std::make_shared<RestPureCompletionQueueImpl>()) {}
 
-void RestCompletionQueueImpl::Run() { tq_->Service(); }
+void RestCompletionQueueImpl::Run() { impl_->Run(); }
 
-void RestCompletionQueueImpl::Shutdown() { tq_->Shutdown(); }
+void RestCompletionQueueImpl::Shutdown() { impl_->Shutdown(); }
 
 void RestCompletionQueueImpl::CancelAll() {}
 
 future<StatusOr<std::chrono::system_clock::time_point>>
 RestCompletionQueueImpl::MakeDeadlineTimer(
     std::chrono::system_clock::time_point deadline) {
-  return tq_->Schedule(deadline);
+  return impl_->MakeDeadlineTimer(deadline);
 }
 
 future<StatusOr<std::chrono::system_clock::time_point>>
 RestCompletionQueueImpl::MakeRelativeTimer(std::chrono::nanoseconds duration) {
-  using std::chrono::system_clock;
-  auto d = std::chrono::duration_cast<system_clock::duration>(duration);
-  if (d < duration) d += system_clock::duration{1};
-  return MakeDeadlineTimer(system_clock::now() + d);
+  return impl_->MakeRelativeTimer(duration);
 }
 
-// Use an "immediately" expiring timer in order to get the thread(s) servicing
-// the TimerQueue to execute the function. However, if the timer
-// expires before .then() is invoked, the lambda will be immediately called and
-// the passing of execution to the queue servicing thread will not occur.
 void RestCompletionQueueImpl::RunAsync(
     std::unique_ptr<internal::RunAsyncBase> function) {
-  ++run_async_counter_;
-  tq_->Schedule([f = std::move(function)](auto) { f->exec(); });
-}
-
-void RestCompletionQueueImpl::StartOperation(
-    std::shared_ptr<internal::AsyncGrpcOperation>,
-    absl::FunctionRef<void(void*)>) {
-  GCP_LOG(FATAL) << " function not supported.\n";
+  impl_->RunAsync(std::move(function));
 }
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/internal/rest_completion_queue_impl.h

@@ -17,6 +17,7 @@
 
 #include "google/cloud/future.h"
 #include "google/cloud/internal/completion_queue_impl.h"
+#include "google/cloud/internal/rest_pure_completion_queue_impl.h"
 #include "google/cloud/internal/timer_queue.h"
 #include "google/cloud/log.h"
 #include "google/cloud/status_or.h"
@@ -69,19 +70,18 @@ class RestCompletionQueueImpl final
   /// This function is not supported by RestCompletionQueueImpl, but as the
   /// function is pure virtual, it must be overridden.
   void StartOperation(std::shared_ptr<internal::AsyncGrpcOperation>,
-                      absl::FunctionRef<void(void*)>) override;
+                      absl::FunctionRef<void(void*)>) override {
+    GCP_LOG(FATAL) << " function not supported.\n";
+  }
 
   /// The underlying gRPC completion queue, which does not exist.
   grpc::CompletionQueue* cq() override { return nullptr; }
 
   /// Some counters for testing and debugging.
-  std::int64_t run_async_counter() const { return run_async_counter_.load(); }
+  std::int64_t run_async_counter() const { return impl_->run_async_counter(); }
 
  private:
-  std::shared_ptr<internal::TimerQueue> tq_;
-
-  // These are metrics used in testing.
-  std::atomic<std::int64_t> run_async_counter_{0};
+  std::shared_ptr<RestPureCompletionQueueImpl> impl_;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/internal/rest_pure_background_threads_impl.cc

@@ -0,0 +1,73 @@
+// Copyright 2026 Google LLC
+//
+// 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
+//
+//     https://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.
+
+#include "google/cloud/internal/rest_pure_background_threads_impl.h"
+#include "google/cloud/future.h"
+#include "google/cloud/internal/call_context.h"
+#include "google/cloud/internal/rest_pure_completion_queue_impl.h"
+#include "google/cloud/log.h"
+#include <algorithm>
+#include <system_error>
+
+namespace google {
+namespace cloud {
+namespace rest_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+AutomaticallyCreatedRestPureBackgroundThreads::
+    AutomaticallyCreatedRestPureBackgroundThreads(std::size_t thread_count)
+    : cq_(std::make_shared<RestPureCompletionQueueImpl>()),
+      pool_(thread_count == 0 ? 1 : thread_count) {
+  std::generate_n(pool_.begin(), pool_.size(), [this] {
+    promise<void> started;
+    auto thread = std::thread(
+        [](RestPureCompletionQueue cq, promise<void>& started,
+           internal::CallContext c) {
+          internal::ScopedCallContext scope(std::move(c));
+          started.set_value();
+          cq.Run();
+        },
+        cq_, std::ref(started), internal::CallContext{});
+    started.get_future().wait();
+    return thread;
+  });
+}
+
+AutomaticallyCreatedRestPureBackgroundThreads::
+    ~AutomaticallyCreatedRestPureBackgroundThreads() {
+  Shutdown();
+}
+
+void AutomaticallyCreatedRestPureBackgroundThreads::Shutdown() {
+  cq_.Shutdown();
+  for (auto& t : pool_) {
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+    try {
+#endif
+      t.join();
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+    } catch (std::system_error const& e) {
+      GCP_LOG(FATAL)
+          << "AutomaticallyCreatedRestPureBackgroundThreads::Shutdown: "
+          << e.what();
+    }
+#endif
+  }
+  pool_.clear();
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace rest_internal
+}  // namespace cloud
+}  // namespace google

google/cloud/internal/rest_pure_background_threads_impl.h

@@ -0,0 +1,62 @@
+// Copyright 2026 Google LLC
+//
+// 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
+//
+//     https://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 GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_REST_PURE_BACKGROUND_THREADS_IMPL_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_REST_PURE_BACKGROUND_THREADS_IMPL_H
+
+#include "google/cloud/internal/rest_pure_completion_queue_impl.h"
+#include "google/cloud/version.h"
+#include <thread>
+#include <vector>
+
+namespace google {
+namespace cloud {
+namespace rest_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+/// This interface mimics `google::cloud::BackgroundThreads` except it returns
+/// a `RestPureCompletionQueue` instead of a `CompletionQueue`. This is
+/// important as `CompletionQueue` has gRPC dependencies we need to avoid in a
+/// pure REST library.
+class RestPureBackgroundThreads {
+ public:
+  virtual ~RestPureBackgroundThreads() = default;
+
+  /// The completion queue used for the background operations.
+  virtual RestPureCompletionQueue cq() const = 0;
+};
+
+/// Background threads that run on a RestPureCompletionQueue.
+class AutomaticallyCreatedRestPureBackgroundThreads
+    : public RestPureBackgroundThreads {
+ public:
+  explicit AutomaticallyCreatedRestPureBackgroundThreads(
+      std::size_t thread_count = 1U);
+  ~AutomaticallyCreatedRestPureBackgroundThreads() override;
+
+  RestPureCompletionQueue cq() const override { return cq_; }
+  void Shutdown();
+  std::size_t pool_size() const { return pool_.size(); }
+
+ private:
+  RestPureCompletionQueue cq_;
+  std::vector<std::thread> pool_;
+};
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace rest_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_REST_PURE_BACKGROUND_THREADS_IMPL_H

google/cloud/internal/rest_pure_completion_queue_impl.cc

@@ -0,0 +1,63 @@
+// Copyright 2026 Google LLC
+//
+// 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
+//
+//     https://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.
+
+#include "google/cloud/internal/rest_pure_completion_queue_impl.h"
+#include "google/cloud/internal/timer_queue.h"
+
+namespace google {
+namespace cloud {
+namespace rest_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+RestPureCompletionQueue::RestPureCompletionQueue()
+    : impl_(std::make_shared<RestPureCompletionQueueImpl>()) {}
+
+RestPureCompletionQueueImpl::RestPureCompletionQueueImpl()
+    : tq_(internal::TimerQueue::Create()) {}
+
+void RestPureCompletionQueueImpl::Run() { tq_->Service(); }
+
+void RestPureCompletionQueueImpl::Shutdown() { tq_->Shutdown(); }
+
+void RestPureCompletionQueueImpl::CancelAll() {}
+
+future<StatusOr<std::chrono::system_clock::time_point>>
+RestPureCompletionQueueImpl::MakeDeadlineTimer(
+    std::chrono::system_clock::time_point deadline) {
+  return tq_->Schedule(deadline);
+}
+
+future<StatusOr<std::chrono::system_clock::time_point>>
+RestPureCompletionQueueImpl::MakeRelativeTimer(
+    std::chrono::nanoseconds duration) {
+  using std::chrono::system_clock;
+  auto d = std::chrono::duration_cast<system_clock::duration>(duration);
+  if (d < duration) d += system_clock::duration{1};
+  return MakeDeadlineTimer(system_clock::now() + d);
+}
+
+// Use an "immediately" expiring timer in order to get the thread(s) servicing
+// the TimerQueue to execute the function. However, if the timer
+// expires before .then() is invoked, the lambda will be immediately called and
+// the passing of execution to the queue servicing thread will not occur.
+void RestPureCompletionQueueImpl::RunAsync(
+    std::unique_ptr<internal::RunAsyncBase> function) {
+  ++run_async_counter_;
+  tq_->Schedule([f = std::move(function)](auto) { f->exec(); });
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace rest_internal
+}  // namespace cloud
+}  // namespace google