Reactive GCS

pom.xml

@@ -9,7 +9,7 @@
 	<name>Spring Framework on Google Cloud</name>
 	<description>Spring Framework on Google Cloud</description>
 	<url>https://spring.io/projects/spring-cloud-gcp</url>
-	<packaging>pom</packaging>
+    <packaging>pom</packaging>
 
 	<scm>
 		<url>https://github.com/GoogleCloudPlatform/spring-cloud-gcp</url>

spring-cloud-gcp-autoconfigure/pom.xml

@@ -330,6 +330,13 @@
             <artifactId>commons-lang3</artifactId>
         </dependency>
 
+        <!-- WebFlux -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <!-- Test -->
         <dependency>
             <groupId>commons-io</groupId>

spring-cloud-gcp-autoconfigure/src/main/java/com/google/cloud/spring/autoconfigure/storage/GcpReactiveStorageAutoConfiguration.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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.
+ */
+
+package com.google.cloud.spring.autoconfigure.storage;
+
+import com.google.api.gax.core.CredentialsProvider;
+import com.google.cloud.spring.autoconfigure.core.GcpProperties;
+import com.google.cloud.spring.core.DefaultCredentialsProvider;
+import com.google.cloud.spring.core.ReactiveTokenProvider;
+import com.google.cloud.spring.storage.GoogleStorageProtocolResolver;
+import com.google.cloud.spring.storage.GoogleStorageTemplate;
+import java.io.IOException;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@AutoConfiguration
+@ConditionalOnClass(WebClient.class)
+@ConditionalOnProperty(value = "spring.cloud.gcp.storage.reactive.enabled", matchIfMissing = true)
+@EnableConfigurationProperties({GcpProperties.class, GcpStorageProperties.class})
+@Import(GoogleStorageProtocolResolver.class)
+public class GcpReactiveStorageAutoConfiguration {
+
+  private final CredentialsProvider credentialsProvider;
+
+  public GcpReactiveStorageAutoConfiguration(
+      CredentialsProvider credentialsProvider,
+      GcpStorageProperties gcpStorageProperties)
+      throws IOException {
+
+    this.credentialsProvider =
+        gcpStorageProperties.getCredentials().hasKey()
+            ? new DefaultCredentialsProvider(gcpStorageProperties)
+            : credentialsProvider;
+  }
+
+  @Bean
+  @ConditionalOnMissingBean
+  public WebClient webClient() {
+    return WebClient.builder().build();
+  }
+
+  @Bean
+  @ConditionalOnMissingBean
+  public ReactiveTokenProvider reactiveTokenProvider(WebClient webClient) throws IOException {
+    return ReactiveTokenProvider.create(credentialsProvider.getCredentials(), webClient);
+  }
+
+  @Bean
+  @ConditionalOnMissingBean
+  public GoogleStorageTemplate storage(WebClient webClient,
+      ReactiveTokenProvider reactiveTokenProvider) {
+    return new GoogleStorageTemplate(webClient, reactiveTokenProvider);
+  }
+
+}

spring-cloud-gcp-core/pom.xml

@@ -26,5 +26,21 @@
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-context</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>com.squareup.okhttp3</groupId>
+			<artifactId>mockwebserver</artifactId>
+			<version>4.10.0</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 </project>

spring-cloud-gcp-core/src/main/java/com/google/auth/oauth2/ComputeEngineTokenProvider.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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.
+ */
+
+package com.google.auth.oauth2;
+
+import static com.google.auth.oauth2.Constants.ACCESS_TOKEN;
+import static com.google.auth.oauth2.Constants.ERROR_PARSING_TOKEN_REFRESH_RESPONSE;
+import static com.google.auth.oauth2.Constants.EXPIRES_IN;
+
+import com.google.api.client.util.GenericData;
+import com.google.cloud.spring.core.ReactiveTokenProvider;
+import java.io.IOException;
+import java.util.Date;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+public class ComputeEngineTokenProvider implements ReactiveTokenProvider {
+
+
+  private final WebClient webClient;
+  private final ComputeEngineCredentials computeEngineCredentials;
+
+  private final String tokenUrl;
+
+  public ComputeEngineTokenProvider(WebClient webClient,
+      ComputeEngineCredentials computeEngineCredentials) {
+    this.webClient = webClient;
+    this.computeEngineCredentials = computeEngineCredentials;
+    tokenUrl = computeEngineCredentials.createTokenUrlWithScopes();
+  }
+
+  public ComputeEngineTokenProvider(WebClient webClient,
+      ComputeEngineCredentials computeEngineCredentials, String tokenUrl) {
+    this.webClient = webClient;
+    this.computeEngineCredentials = computeEngineCredentials;
+    this.tokenUrl = tokenUrl;
+  }
+
+  @Override
+  public Mono<AccessToken> retrieve() {
+    return webClient.get().uri(tokenUrl)
+        .header("Metadata-Flavor", "Google")
+        .retrieve()
+        .bodyToMono(GenericData.class)
+        .flatMap(gd -> {
+          try {
+            String tokenValue = OAuth2Utils.validateString(gd, ACCESS_TOKEN,
+                ERROR_PARSING_TOKEN_REFRESH_RESPONSE);
+            int expiresInSeconds = OAuth2Utils.validateInt32(gd, EXPIRES_IN,
+                ERROR_PARSING_TOKEN_REFRESH_RESPONSE);
+            long expiresAtMilliseconds =
+                computeEngineCredentials.clock.currentTimeMillis() + (expiresInSeconds * 1000L);
+            AccessToken accessToken = new AccessToken(tokenValue, new Date(expiresAtMilliseconds));
+            return Mono.just(accessToken);
+          } catch (IOException e) {
+            return Mono.error(e);
+          }
+        });
+  }
+
+}

spring-cloud-gcp-core/src/main/java/com/google/auth/oauth2/Constants.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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.
+ */
+
+package com.google.auth.oauth2;
+
+public class Constants {
+
+  static final String ACCESS_TOKEN = "access_token";
+  static final String EXPIRES_IN = "expires_in";
+  static final String ERROR_PARSING_TOKEN_REFRESH_RESPONSE = "Error parsing token refresh response. ";
+
+  private Constants() {
+  }
+
+}

spring-cloud-gcp-core/src/main/java/com/google/auth/oauth2/ServiceAccountTokenProvider.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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.
+ */
+
+package com.google.auth.oauth2;
+
+import com.google.api.client.json.JsonFactory;
+import com.google.api.client.util.GenericData;
+import com.google.cloud.spring.core.ReactiveTokenProvider;
+import java.io.IOException;
+import java.util.Date;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+public class ServiceAccountTokenProvider implements ReactiveTokenProvider {
+
+  private static final String GRANT_TYPE = "grant_type";
+  private static final String ASSERTION = "assertion";
+  private final WebClient webClient;
+  private final ServiceAccountCredentials serviceAccountCredentials;
+
+  private final String tokenUrl;
+
+  public ServiceAccountTokenProvider(WebClient webClient,
+      ServiceAccountCredentials serviceAccountCredentials) {
+    this.webClient = webClient;
+    this.serviceAccountCredentials = serviceAccountCredentials;
+    tokenUrl = OAuth2Utils.TOKEN_SERVER_URI.toString();
+  }
+
+  public ServiceAccountTokenProvider(WebClient webClient,
+      ServiceAccountCredentials serviceAccountCredentials, String tokenUrl) {
+    this.webClient = webClient;
+    this.serviceAccountCredentials = serviceAccountCredentials;
+    this.tokenUrl = tokenUrl;
+  }
+
+  @Override
+  public Mono<AccessToken> retrieve() {
+    JsonFactory jsonFactory = OAuth2Utils.JSON_FACTORY;
+    long currentTime = serviceAccountCredentials.clock.currentTimeMillis();
+
+    try {
+      String assertion = serviceAccountCredentials.createAssertion(jsonFactory, currentTime);
+
+      return webClient.post()
+          .uri(tokenUrl)
+          .contentType(MediaType.APPLICATION_FORM_URLENCODED)
+          .body(refreshForm(assertion))
+          .retrieve()
+          .bodyToMono(GenericData.class)
+          .flatMap(gd -> {
+            try {
+              AccessToken accessToken = getAccessToken(gd);
+              return Mono.just(accessToken);
+            } catch (IOException e) {
+              return Mono.error(e);
+            }
+          });
+
+    } catch (IOException e) {
+      return Mono.error(e);
+    }
+  }
+
+  private AccessToken getAccessToken(GenericData gd) throws IOException {
+    String tokenValue = OAuth2Utils.validateString(gd, "access_token",
+        "Error parsing token refresh response. ");
+    int expiresInSeconds = OAuth2Utils.validateInt32(gd, "expires_in",
+        "Error parsing token refresh response. ");
+    long expiresAtMilliseconds =
+        serviceAccountCredentials.clock.currentTimeMillis() + (long) expiresInSeconds * 1000L;
+    AccessToken accessToken = new AccessToken(tokenValue, new Date(expiresAtMilliseconds));
+    return accessToken;
+  }
+
+  private static BodyInserters.FormInserter<String> refreshForm(String assertion) {
+    return BodyInserters.fromFormData(GRANT_TYPE, OAuth2Utils.GRANT_TYPE_JWT_BEARER)
+        .with(ASSERTION, assertion);
+  }
+
+}