implement Oauth2TokenFetcher

src/core/lib/security/credentials/oauth2/oauth2_credentials.cc

@@ -145,12 +145,6 @@ void grpc_auth_refresh_token_destruct(grpc_auth_refresh_token* refresh_token) {
 // Oauth2 Token Fetcher credentials.
 //
 
-grpc_oauth2_token_fetcher_credentials::
-    ~grpc_oauth2_token_fetcher_credentials() {
-  gpr_mu_destroy(&mu_);
-  grpc_pollset_set_destroy(grpc_polling_entity_pollset_set(&pollent_));
-}
-
 grpc_credentials_status
 grpc_oauth2_token_fetcher_credentials_parse_server_response(
     const grpc_http_response* response,
@@ -234,6 +228,90 @@ end:
   return status;
 }
 
+namespace grpc_core {
+
+namespace {
+
+// An oauth2 token.
+class Oauth2Token : public TokenFetcherCredentials::Token {
+ public:
+  Oauth2Token(Slice token, Timestamp expiration)
+      : token_(std::move(token)), expiration_(expiration) {}
+
+  Timestamp ExpirationTime() override { return expiration_; }
+
+  void AddTokenToClientInitialMetadata(ClientMetadata& metadata) override {
+    metadata.Append(
+        GRPC_AUTHORIZATION_METADATA_KEY, token_.Ref(),
+        [](absl::string_view, const grpc_core::Slice&) { abort(); });
+  }
+
+ private:
+  Slice token_;
+  Timestamp expiration_;
+};
+
+// State held for a pending HTTP request.
+struct HttpRequestState {
+  explicit HttpRequestState(
+      absl::AnyInvocable<void(
+          absl::StatusOr<RefCountedPtr<TokenFetcherCredentials::Token>>)>
+          on_done)
+      : on_done(std::move(on_done)) {
+    GRPC_CLOSURE_INIT(&on_http_response, OnHttpResponse, this, nullptr);
+  }
+  ~HttpRequestState() { grpc_http_response_destroy(&response); }
+
+  static void OnHttpResponse(void* arg, grpc_error_handle error) {
+    std::unique_ptr<HttpRequestState> self(static_cast<HttpRequestState*>(arg));
+    if (!error.ok()) {
+      self->on_done(std::move(error));
+      return;
+    }
+    // Parse oauth2 token.
+    absl::optional<Slice> access_token_value;
+    Duration token_lifetime;
+    grpc_credentials_status status =
+        grpc_oauth2_token_fetcher_credentials_parse_server_response(
+            &self->response, &access_token_value, &token_lifetime);
+    if (status != GRPC_CREDENTIALS_OK) {
+      self->on_done(
+          absl::UnavailableError("error parsing oauth2 token"));
+      return;
+    }
+    self->on_done(
+        MakeRefCounted<Oauth2Token>(std::move(*access_token_value),
+                                    Timestamp::Now() + token_lifetime));
+  }
+
+  absl::AnyInvocable<void(
+      absl::StatusOr<RefCountedPtr<TokenFetcherCredentials::Token>>)> on_done;
+  grpc_closure on_http_response;
+  grpc_http_response response;
+};
+
+}  // namespace
+
+OrphanablePtr<HttpRequest> Oauth2TokenFetcher::FetchToken(
+    grpc_polling_entity* pollent, Timestamp deadline,
+    absl::AnyInvocable<void(
+        absl::StatusOr<RefCountedPtr<TokenFetcherCredentials::Token>>)>
+        on_done) {
+  auto* state = new HttpRequestState(std::move(on_done));
+  OrphanablePtr<HttpRequest> http_request = ConstructHttpRequest(
+      pollent, deadline, &state->response, &state->on_http_response);
+  http_request->Start();
+  return http_request;
+}
+
+}  // namespace grpc_core
+
+grpc_oauth2_token_fetcher_credentials::
+    ~grpc_oauth2_token_fetcher_credentials() {
+  gpr_mu_destroy(&mu_);
+  grpc_pollset_set_destroy(grpc_polling_entity_pollset_set(&pollent_));
+}
+
 static void on_oauth2_token_fetcher_http_response(void* user_data,
                                                   grpc_error_handle error) {
   GRPC_LOG_IF_ERROR("oauth_fetch", error);

src/core/lib/security/credentials/oauth2/oauth2_credentials.h

@@ -43,6 +43,7 @@
 #include "src/core/lib/promise/activity.h"
 #include "src/core/lib/promise/arena_promise.h"
 #include "src/core/lib/security/credentials/credentials.h"
+#include "src/core/lib/security/credentials/token_fetcher/token_fetcher_credentials.h"
 #include "src/core/lib/slice/slice.h"
 #include "src/core/lib/transport/transport.h"
 #include "src/core/lib/uri/uri_parser.h"
@@ -109,6 +110,25 @@ struct grpc_oauth2_pending_get_request_metadata
 //  This object is a base for credentials that need to acquire an oauth2 token
 //  from an http service.
 
+namespace grpc_core {
+
+// A base class for oauth2 token fetching.
+// Subclasses must implement ConstructHttpRequest().
+class Oauth2TokenFetcher : public TokenFetcherCredentials::TokenFetcher {
+ public:
+  OrphanablePtr<HttpRequest> FetchToken(
+        grpc_polling_entity* pollent, Timestamp deadline,
+        absl::AnyInvocable<void(
+            absl::StatusOr<RefCountedPtr<TokenFetcherCredentials::Token>>)>
+            on_done) final;
+
+  virtual OrphanablePtr<HttpRequest> ConstructHttpRequest(
+        grpc_polling_entity* pollent, Timestamp deadline,
+        grpc_http_response* response, grpc_closure* on_complete) = 0;
+};
+
+}  // namespace grpc_core
+
 class grpc_oauth2_token_fetcher_credentials : public grpc_call_credentials {
  public:
   grpc_oauth2_token_fetcher_credentials();

src/core/lib/security/credentials/token_fetcher/token_fetcher_credentials.h

@@ -41,6 +41,7 @@
 namespace grpc_core {
 
 // A base class for credentials that fetch tokens via an HTTP request.
+// Subclasses must implement TokenFetcher.
 class TokenFetcherCredentials : public grpc_call_credentials {
  public:
   // Represents a token.
@@ -62,6 +63,7 @@ class TokenFetcherCredentials : public grpc_call_credentials {
 
     // Fetches a token.  The on_done callback will be invoked when complete.
     // The fetch may be cancelled by orphaning the returned HttpRequest.
+    // The on_done callback will be invoked even when cancelled.
     virtual OrphanablePtr<HttpRequest> FetchToken(
         grpc_polling_entity* pollent, Timestamp deadline,
         absl::AnyInvocable<void(absl::StatusOr<RefCountedPtr<Token>>)>
@@ -97,8 +99,10 @@ class TokenFetcherCredentials : public grpc_call_credentials {
   const std::unique_ptr<TokenFetcher> token_fetcher_;
 
   Mutex mu_;
+  // Either the cached token or a pending request to fetch the token.
   absl::variant<RefCountedPtr<Token>, OrphanablePtr<HttpRequest>> token_
       ABSL_GUARDED_BY(&mu_);
+  // Calls that are queued up waiting for the token.
   absl::flat_hash_set<RefCountedPtr<PendingCall>> pending_calls_
       ABSL_GUARDED_BY(&mu_);
   grpc_polling_entity pollent_ ABSL_GUARDED_BY(&mu_);