feat(gax): add ResumableUploadCallable creation to Callables and HttpJsonCallableFactory

sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallableFactory.java

@@ -29,8 +29,10 @@
  */
 package com.google.api.gax.httpjson;
 
+import com.google.api.core.BetaApi;
 import com.google.api.core.ObsoleteApi;
 import com.google.api.gax.longrunning.OperationSnapshot;
+import com.google.api.gax.resumable.ResumableUploadClient;
 import com.google.api.gax.rpc.BatchingCallSettings;
 import com.google.api.gax.rpc.Callables;
 import com.google.api.gax.rpc.ClientContext;
@@ -39,6 +41,8 @@
 import com.google.api.gax.rpc.OperationCallSettings;
 import com.google.api.gax.rpc.OperationCallable;
 import com.google.api.gax.rpc.PagedCallSettings;
+import com.google.api.gax.rpc.ResumableUploadCallSettings;
+import com.google.api.gax.rpc.ResumableUploadCallable;
 import com.google.api.gax.rpc.ServerStreamingCallSettings;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.api.gax.rpc.UnaryCallSettings;
@@ -221,6 +225,27 @@
     return callable.withDefaultCallContext(clientContext.getDefaultCallContext());
   }
 
+  /**
+   * Creates a {@link ResumableUploadCallable} to execute resumable uploads. Designed for use by
+   * generated code.
+   *
+   * @param httpJsonCallSettings the http/json call settings
+   * @param callSettings settings configuring chunk size
+   * @param clientContext client context providing default call context
+   * @return {@link ResumableUploadCallable} callable object
+   */
+  @BetaApi
+  public static <RequestT, ResponseT>
+      ResumableUploadCallable<RequestT, ResponseT> createResumableUploadCallable(
+          HttpJsonCallSettings<RequestT, ResponseT> httpJsonCallSettings,
+          ResumableUploadCallSettings callSettings,
+          ClientContext clientContext) {
+    ResumableUploadClient<RequestT, ResponseT> uploadClient =
+        HttpJsonResumableUploadClient.create(
+            clientContext, httpJsonCallSettings.getMethodDescriptor());
+    return Callables.resumableUpload(uploadClient, callSettings, clientContext);
+  }
+
   static ApiTracerContext getApiTracerContext(ApiMethodDescriptor<?, ?> methodDescriptor) {
     return ApiTracerContext.newBuilder()
         .setFullMethodName(methodDescriptor.getFullMethodName())

sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonCallableFactoryTest.java

@@ -34,6 +34,9 @@
 import static org.mockito.Mockito.mock;
 
 import com.google.api.client.http.HttpMethods;
+import com.google.api.gax.rpc.ClientContext;
+import com.google.api.gax.rpc.ResumableUploadCallSettings;
+import com.google.api.gax.rpc.ResumableUploadCallable;
 import com.google.api.gax.tracing.ApiTracerContext;
 import com.google.api.gax.tracing.SpanName;
 import com.google.api.pathtemplate.PathTemplate;
@@ -107,4 +110,34 @@ void testGetApiTracerContextInvalid() {
       assertThat(actualError).isNotNull();
     }
   }
+
+  @Test
+  void testCreateResumableUploadCallable() {
+    @SuppressWarnings("unchecked")
+    ApiMethodDescriptor<String, String> descriptor =
+        ApiMethodDescriptor.<String, String>newBuilder()
+            .setFullMethodName("test/upload")
+            .setHttpMethod(HttpMethods.POST)
+            .setRequestFormatter(createMockRequestFormatter())
+            .setResponseParser(
+                mock(HttpResponseParser.class, Mockito.withSettings().withoutAnnotations()))
+            .build();
+
+    HttpJsonCallSettings<String, String> httpJsonCallSettings =
+        HttpJsonCallSettings.<String, String>newBuilder().setMethodDescriptor(descriptor).build();
+
+    ResumableUploadCallSettings callSettings =
+        ResumableUploadCallSettings.newBuilder().setChunkSize(256 * 1024).build();
+
+    ClientContext clientContext =
+        ClientContext.newBuilder()
+            .setDefaultCallContext(HttpJsonCallContext.createDefault())
+            .build();
+
+    ResumableUploadCallable<String, String> callable =
+        HttpJsonCallableFactory.createResumableUploadCallable(
+            httpJsonCallSettings, callSettings, clientContext);
+
+    assertThat(callable).isNotNull();
+  }
 }

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/Callables.java

@@ -33,6 +33,7 @@
 import com.google.api.gax.batching.BatchingSettings;
 import com.google.api.gax.longrunning.OperationResponsePollAlgorithm;
 import com.google.api.gax.longrunning.OperationSnapshot;
+import com.google.api.gax.resumable.ResumableUploadClient;
 import com.google.api.gax.retrying.ExponentialRetryAlgorithm;
 import com.google.api.gax.retrying.RetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
@@ -270,6 +271,26 @@
         initialCallable, scheduler, longRunningClient, operationCallSettings);
   }
 
+  /**
+   * Creates a {@link ResumableUploadCallable} to execute resumable uploads. Designed for use by
+   * generated code.
+   *
+   * @param uploadClient client executing the wire-level upload protocol
+   * @param callSettings settings configuring chunk size
+   * @param clientContext client context providing default call context and executor
+   * @return {@link ResumableUploadCallable} callable object
+   */
+  public static <RequestT, ResponseT> ResumableUploadCallable<RequestT, ResponseT> resumableUpload(
+      ResumableUploadClient<RequestT, ResponseT> uploadClient,
+      ResumableUploadCallSettings callSettings,
+      ClientContext clientContext) {
+    return new ResumableUploadCallableImpl<>(
+        uploadClient,
+        callSettings,
+        clientContext.getDefaultCallContext(),
+        clientContext.getExecutor());
+  }
+
   private static boolean areRetriesDisabled(
       Collection<StatusCode.Code> retryableCodes, RetrySettings retrySettings) {
     return retrySettings.getMaxAttempts() == 1

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/CallableTest.java

@@ -30,6 +30,7 @@
 package com.google.api.gax.rpc;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -42,6 +43,7 @@
 import com.google.api.core.ApiClock;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.resumable.ResumableUploadClient;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.testing.FakeCallContext;
 import java.util.concurrent.ScheduledExecutorService;
@@ -207,4 +209,21 @@ void testWatched_usesJavaTimeMethods() {
     verify(callContext, atLeastOnce()).withStreamIdleTimeoutDuration(eq(timeout));
     verify(callContext, atLeastOnce()).withStreamWaitTimeoutDuration(eq(timeout));
   }
+
+  @Test
+  void testResumableUploadCallable() {
+    @SuppressWarnings("unchecked")
+    ResumableUploadClient<String, String> uploadClient =
+        mock(ResumableUploadClient.class, Mockito.withSettings().withoutAnnotations());
+    ResumableUploadCallSettings settings =
+        ResumableUploadCallSettings.newBuilder().setChunkSize(1024).build();
+    ScheduledExecutorService executor =
+        mock(ScheduledExecutorService.class, Mockito.withSettings().withoutAnnotations());
+    ClientContext context = clientContext.toBuilder().setExecutor(executor).build();
+
+    ResumableUploadCallable<String, String> callable =
+        Callables.resumableUpload(uploadClient, settings, context);
+
+    assertNotNull(callable);
+  }
 }