feat(pubsub): implement GUAC for TopicAdmin

google/cloud/pubsub/CMakeLists.txt

@@ -58,6 +58,8 @@ add_library(
     internal/non_constructible.h
     internal/ordering_key_publisher_connection.cc
     internal/ordering_key_publisher_connection.h
+    internal/publisher_auth.cc
+    internal/publisher_auth.h
     internal/publisher_logging.cc
     internal/publisher_logging.h
     internal/publisher_metadata.cc
@@ -236,6 +238,7 @@ function (google_cloud_cpp_pubsub_client_define_tests)
         internal/defaults_test.cc
         internal/flow_controlled_publisher_connection_test.cc
         internal/ordering_key_publisher_connection_test.cc
+        internal/publisher_auth_test.cc
         internal/publisher_logging_test.cc
         internal/publisher_metadata_test.cc
         internal/publisher_round_robin_test.cc

google/cloud/pubsub/google_cloud_cpp_pubsub.bzl

@@ -29,6 +29,7 @@ google_cloud_cpp_pubsub_hdrs = [
     "internal/flow_controlled_publisher_connection.h",
     "internal/non_constructible.h",
     "internal/ordering_key_publisher_connection.h",
+    "internal/publisher_auth.h",
     "internal/publisher_logging.h",
     "internal/publisher_metadata.h",
     "internal/publisher_round_robin.h",
@@ -85,6 +86,7 @@ google_cloud_cpp_pubsub_srcs = [
     "internal/defaults.cc",
     "internal/flow_controlled_publisher_connection.cc",
     "internal/ordering_key_publisher_connection.cc",
+    "internal/publisher_auth.cc",
     "internal/publisher_logging.cc",
     "internal/publisher_metadata.cc",
     "internal/publisher_round_robin.cc",

google/cloud/pubsub/integration_tests/topic_admin_integration_test.cc

@@ -16,6 +16,7 @@
 #include "google/cloud/pubsub/testing/test_retry_policies.h"
 #include "google/cloud/pubsub/topic_admin_client.h"
 #include "google/cloud/pubsub/version.h"
+#include "google/cloud/credentials.h"
 #include "google/cloud/internal/getenv.h"
 #include "google/cloud/internal/random.h"
 #include "google/cloud/testing_util/integration_test.h"
@@ -37,6 +38,7 @@ using ::google::cloud::testing_util::ScopedEnvironment;
 using ::google::cloud::testing_util::StatusIs;
 using ::testing::AnyOf;
 using ::testing::Contains;
+using ::testing::IsEmpty;
 using ::testing::Not;
 
 bool UsingEmulator() {
@@ -50,34 +52,35 @@ TopicAdminClient MakeTestTopicAdminClient() {
 using TopicAdminIntegrationTest =
     ::google::cloud::testing_util::IntegrationTest;
 
+StatusOr<std::vector<std::string>> TopicNames(TopicAdminClient client,
+                                              std::string const& project_id) {
+  std::vector<std::string> names;
+  for (auto& topic : client.ListTopics(project_id)) {
+    if (!topic) return std::move(topic).status();
+    names.push_back(std::move(*topic->mutable_name()));
+  }
+  return names;
+}
+
 TEST_F(TopicAdminIntegrationTest, TopicCRUD) {
   auto project_id =
       google::cloud::internal::GetEnv("GOOGLE_CLOUD_PROJECT").value_or("");
-  ASSERT_FALSE(project_id.empty());
-
-  auto topic_names = [](TopicAdminClient client,
-                        std::string const& project_id) {
-    std::vector<std::string> names;
-    for (auto& topic : client.ListTopics(project_id)) {
-      EXPECT_STATUS_OK(topic);
-      if (!topic) break;
-      names.push_back(topic->name());
-    }
-    return names;
-  };
+  ASSERT_THAT(project_id, Not(IsEmpty()));
 
   auto generator = google::cloud::internal::MakeDefaultPRNG();
   Topic topic(project_id, pubsub_testing::RandomTopicId(generator));
 
   auto publisher = TopicAdminClient(MakeTopicAdminConnection());
-
-  EXPECT_THAT(topic_names(publisher, project_id),
-              Not(Contains(topic.FullName())));
+  auto names = TopicNames(publisher, project_id);
+  ASSERT_STATUS_OK(names);
+  EXPECT_THAT(*names, Not(Contains(topic.FullName())));
 
   auto create_response = publisher.CreateTopic(TopicBuilder(topic));
   ASSERT_THAT(create_response,
               AnyOf(IsOk(), StatusIs(StatusCode::kAlreadyExists)));
-  EXPECT_THAT(topic_names(publisher, project_id), Contains(topic.FullName()));
+  names = TopicNames(publisher, project_id);
+  ASSERT_STATUS_OK(names);
+  EXPECT_THAT(*names, Contains(topic.FullName()));
 
   auto get_response = publisher.GetTopic(topic);
   ASSERT_STATUS_OK(get_response);
@@ -98,8 +101,20 @@ TEST_F(TopicAdminIntegrationTest, TopicCRUD) {
 
   auto delete_response = publisher.DeleteTopic(topic);
   EXPECT_THAT(delete_response, AnyOf(IsOk(), StatusIs(StatusCode::kNotFound)));
-  EXPECT_THAT(topic_names(publisher, project_id),
-              Not(Contains(topic.FullName())));
+  names = TopicNames(publisher, project_id);
+  ASSERT_STATUS_OK(names);
+  EXPECT_THAT(*names, Not(Contains(topic.FullName())));
+}
+
+TEST_F(TopicAdminIntegrationTest, UnifiedCredentials) {
+  auto project_id =
+      google::cloud::internal::GetEnv("GOOGLE_CLOUD_PROJECT").value_or("");
+  ASSERT_THAT(project_id, Not(IsEmpty()));
+  auto credentials = UsingEmulator() ? MakeInsecureCredentials()
+                                     : MakeGoogleDefaultCredentials();
+  auto client = TopicAdminClient(MakeTopicAdminConnection(
+      Options{}.set<UnifiedCredentialsOption>(std::move(credentials))));
+  ASSERT_STATUS_OK(TopicNames(client, project_id));
 }
 
 TEST_F(TopicAdminIntegrationTest, CreateTopicFailure) {

google/cloud/pubsub/internal/publisher_auth.cc

@@ -0,0 +1,110 @@
+// Copyright 2021 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
+//
+//     http://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/pubsub/internal/publisher_auth.h"
+#include "google/cloud/internal/log_wrapper.h"
+
+namespace google {
+namespace cloud {
+namespace pubsub_internal {
+inline namespace GOOGLE_CLOUD_CPP_PUBSUB_NS {
+
+StatusOr<google::pubsub::v1::Topic> PublisherAuth::CreateTopic(
+    grpc::ClientContext& context, google::pubsub::v1::Topic const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->CreateTopic(context, request);
+}
+
+StatusOr<google::pubsub::v1::Topic> PublisherAuth::GetTopic(
+    grpc::ClientContext& context,
+    google::pubsub::v1::GetTopicRequest const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->GetTopic(context, request);
+}
+
+StatusOr<google::pubsub::v1::Topic> PublisherAuth::UpdateTopic(
+    grpc::ClientContext& context,
+    google::pubsub::v1::UpdateTopicRequest const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->UpdateTopic(context, request);
+}
+
+StatusOr<google::pubsub::v1::ListTopicsResponse> PublisherAuth::ListTopics(
+    grpc::ClientContext& context,
+    google::pubsub::v1::ListTopicsRequest const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->ListTopics(context, request);
+}
+
+Status PublisherAuth::DeleteTopic(
+    grpc::ClientContext& context,
+    google::pubsub::v1::DeleteTopicRequest const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->DeleteTopic(context, request);
+}
+
+StatusOr<google::pubsub::v1::DetachSubscriptionResponse>
+PublisherAuth::DetachSubscription(
+    grpc::ClientContext& context,
+    google::pubsub::v1::DetachSubscriptionRequest const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->DetachSubscription(context, request);
+}
+
+StatusOr<google::pubsub::v1::ListTopicSubscriptionsResponse>
+PublisherAuth::ListTopicSubscriptions(
+    grpc::ClientContext& context,
+    google::pubsub::v1::ListTopicSubscriptionsRequest const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->ListTopicSubscriptions(context, request);
+}
+
+StatusOr<google::pubsub::v1::ListTopicSnapshotsResponse>
+PublisherAuth::ListTopicSnapshots(
+    grpc::ClientContext& context,
+    google::pubsub::v1::ListTopicSnapshotsRequest const& request) {
+  auto status = auth_->ConfigureContext(context);
+  if (!status.ok()) return status;
+  return child_->ListTopicSnapshots(context, request);
+}
+
+future<StatusOr<google::pubsub::v1::PublishResponse>>
+PublisherAuth::AsyncPublish(google::cloud::CompletionQueue& cq,
+                            std::unique_ptr<grpc::ClientContext> context,
+                            google::pubsub::v1::PublishRequest const& request) {
+  using ReturnType = StatusOr<google::pubsub::v1::PublishResponse>;
+  auto child = child_;
+  return auth_->AsyncConfigureContext(std::move(context))
+      .then([cq, child,
+             request](future<StatusOr<std::unique_ptr<grpc::ClientContext>>>
+                          f) mutable {
+        auto context = f.get();
+        if (!context) {
+          return make_ready_future(ReturnType(std::move(context).status()));
+        }
+        return child->AsyncPublish(cq, *std::move(context), request);
+      });
+}
+
+}  // namespace GOOGLE_CLOUD_CPP_PUBSUB_NS
+}  // namespace pubsub_internal
+}  // namespace cloud
+}  // namespace google

google/cloud/pubsub/internal/publisher_auth.h

@@ -0,0 +1,84 @@
+// Copyright 2021 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
+//
+//     http://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_PUBSUB_INTERNAL_PUBLISHER_AUTH_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_PUBSUB_INTERNAL_PUBLISHER_AUTH_H
+
+#include "google/cloud/pubsub/internal/publisher_stub.h"
+#include "google/cloud/pubsub/version.h"
+#include "google/cloud/internal/unified_grpc_credentials.h"
+#include <memory>
+
+namespace google {
+namespace cloud {
+namespace pubsub_internal {
+inline namespace GOOGLE_CLOUD_CPP_PUBSUB_NS {
+
+class PublisherAuth : public PublisherStub {
+ public:
+  PublisherAuth(
+      std::shared_ptr<google::cloud::internal::GrpcAuthenticationStrategy> auth,
+      std::shared_ptr<PublisherStub> child)
+      : auth_(std::move(auth)), child_(std::move(child)) {}
+
+  StatusOr<google::pubsub::v1::Topic> CreateTopic(
+      grpc::ClientContext& context,
+      google::pubsub::v1::Topic const& request) override;
+
+  StatusOr<google::pubsub::v1::Topic> GetTopic(
+      grpc::ClientContext& context,
+      google::pubsub::v1::GetTopicRequest const& request) override;
+
+  StatusOr<google::pubsub::v1::Topic> UpdateTopic(
+      grpc::ClientContext& context,
+      google::pubsub::v1::UpdateTopicRequest const& request) override;
+
+  StatusOr<google::pubsub::v1::ListTopicsResponse> ListTopics(
+      grpc::ClientContext& context,
+      google::pubsub::v1::ListTopicsRequest const& request) override;
+
+  Status DeleteTopic(
+      grpc::ClientContext& context,
+      google::pubsub::v1::DeleteTopicRequest const& request) override;
+
+  StatusOr<google::pubsub::v1::DetachSubscriptionResponse> DetachSubscription(
+      grpc::ClientContext& context,
+      google::pubsub::v1::DetachSubscriptionRequest const& request) override;
+
+  StatusOr<google::pubsub::v1::ListTopicSubscriptionsResponse>
+  ListTopicSubscriptions(
+      grpc::ClientContext& context,
+      google::pubsub::v1::ListTopicSubscriptionsRequest const& request)
+      override;
+
+  StatusOr<google::pubsub::v1::ListTopicSnapshotsResponse> ListTopicSnapshots(
+      grpc::ClientContext& context,
+      google::pubsub::v1::ListTopicSnapshotsRequest const& request) override;
+
+  future<StatusOr<google::pubsub::v1::PublishResponse>> AsyncPublish(
+      google::cloud::CompletionQueue& cq,
+      std::unique_ptr<grpc::ClientContext> context,
+      google::pubsub::v1::PublishRequest const& request) override;
+
+ private:
+  std::shared_ptr<google::cloud::internal::GrpcAuthenticationStrategy> auth_;
+  std::shared_ptr<PublisherStub> child_;
+};
+
+}  // namespace GOOGLE_CLOUD_CPP_PUBSUB_NS
+}  // namespace pubsub_internal
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_PUBSUB_INTERNAL_PUBLISHER_AUTH_H