diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java index 8d4e441853dd..3ad03a2b535a 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java @@ -48,7 +48,7 @@ public abstract class ResumableUploadCallable { protected ResumableUploadCallable() {} /** - * Performs a new resumable upload asynchronously. + * Performs a new resumable upload asynchronously with default call context and default settings. * *

The provided {@code payload} stream is consumed asynchronously by the returned {@link * ResumableUploadFuture} and will be closed automatically upon completion, failure, or @@ -56,11 +56,67 @@ protected ResumableUploadCallable() {} * * @param request the request message * @param payload the data payload input stream to upload and close - * @param settings call settings overrides; may be {@code null} + * @return future for tracking and controlling the upload + */ + public ResumableUploadFuture futureCall(RequestT request, InputStream payload) { + return futureCall(request, payload, null, null); + } + + /** + * Performs a new resumable upload asynchronously with a call context override and default + * settings. + * + *

The provided {@code payload} stream is consumed asynchronously by the returned {@link + * ResumableUploadFuture} and will be closed automatically upon completion, failure, or + * cancellation. + * + * @param request the request message + * @param payload the data payload input stream to upload and close + * @param context call context overrides (e.g. extra headers, credentials, timeout); may be {@code + * null} + * @return future for tracking and controlling the upload + */ + public ResumableUploadFuture futureCall( + RequestT request, InputStream payload, @Nullable ApiCallContext context) { + return futureCall(request, payload, context, null); + } + + /** + * Performs a new resumable upload asynchronously with settings overrides and default call + * context. + * + *

The provided {@code payload} stream is consumed asynchronously by the returned {@link + * ResumableUploadFuture} and will be closed automatically upon completion, failure, or + * cancellation. + * + * @param request the request message + * @param payload the data payload input stream to upload and close + * @param settings request-level call settings overrides; may be {@code null} + * @return future for tracking and controlling the upload + */ + public ResumableUploadFuture futureCall( + RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) { + return futureCall(request, payload, null, settings); + } + + /** + * Performs a new resumable upload asynchronously with call context and settings overrides. + * + *

The provided {@code payload} stream is consumed asynchronously by the returned {@link + * ResumableUploadFuture} and will be closed automatically upon completion, failure, or + * cancellation. + * + * @param request the request message + * @param payload the data payload input stream to upload and close + * @param context call context overrides; may be {@code null} + * @param settings request-level call settings overrides; may be {@code null} * @return future for tracking and controlling the upload */ public abstract ResumableUploadFuture futureCall( - RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings); + RequestT request, + InputStream payload, + @Nullable ApiCallContext context, + @Nullable ResumableUploadCallSettings settings); /** * Resumes an existing resumable upload session asynchronously using a saved session URL. diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java index aca5d1684287..2bd063fe9426 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java @@ -75,14 +75,18 @@ public ResumableUploadCallableImpl( @Override public ResumableUploadFuture futureCall( - RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) { + RequestT request, + InputStream payload, + @Nullable ApiCallContext context, + @Nullable ResumableUploadCallSettings settings) { checkNotNull(request, "request must not be null"); checkNotNull(payload, "payload must not be null"); ResumableUploadCallSettings effectiveSettings = defaultCallSettings.merge(settings); + ApiCallContext effectiveCallContext = defaultCallContext.merge(context); ApiFuture startFuture; try { - startFuture = client.startUploadCallable().futureCall(request, defaultCallContext); + startFuture = client.startUploadCallable().futureCall(request, effectiveCallContext); } catch (Throwable t) { startFuture = ApiFutures.immediateFailedFuture(t); } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java index f683fcb638d0..79e7675c4114 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java @@ -112,7 +112,7 @@ void testUploadCallable_singleChunk_happyPath() throws Exception { ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-single"))); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("hello"), null); + callable.futureCall("resource-path", streamOf("hello")); assertThat(future.get()).isEqualTo("response-single"); assertThat(future.isDone()).isTrue(); @@ -136,7 +136,7 @@ void testUploadCallable_multiChunk_happyPath() throws Exception { .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-multi"))); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("01234567890123456789"), null); + callable.futureCall("resource-path", streamOf("01234567890123456789")); assertThat(future.get()).isEqualTo("response-multi"); assertThat(future.isDone()).isTrue(); @@ -156,7 +156,7 @@ void testUploadCallable_zeroByteUpload_finalizesSuccessfully() throws Exception .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-zero"))); ResumableUploadFuture future = - callable.futureCall("resource-path", new ByteArrayInputStream(new byte[0]), null); + callable.futureCall("resource-path", new ByteArrayInputStream(new byte[0])); assertThat(future.get()).isEqualTo("response-zero"); @@ -176,7 +176,7 @@ void testUploadCallable_singleChunkWithSeparateZeroByteFinalize_completesSuccess ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-exact-single"))); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("12345678"), null); + callable.futureCall("resource-path", streamOf("12345678")); assertThat(future.get()).isEqualTo("response-exact-single"); @@ -194,7 +194,7 @@ void testUploadCallable_nullResponse_completesSuccessfully() throws Exception { .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, null))); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + callable.futureCall("resource-path", streamOf("data")); assertThat(future.get()).isNull(); assertThat(future.isDone()).isTrue(); @@ -217,7 +217,7 @@ void testUploadCallable_cancelInFlight_haltsUpload() throws Exception { }); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("0123456789ABCDEF"), null); + callable.futureCall("resource-path", streamOf("0123456789ABCDEF")); assertThat(chunkStarted.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(future.cancel(true)).isTrue(); @@ -233,7 +233,7 @@ void testUploadCallable_setInFlightFutureAfterCancel_immediatelyCancelsFuture() SettableApiFuture startFuture = SettableApiFuture.create(); when(mockStartCallable.futureCall(any(), any())).thenReturn(startFuture); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + callable.futureCall("resource-path", streamOf("data")); assertThat(future.cancel(true)).isTrue(); assertThat(future.isCancelled()).isTrue(); @@ -248,7 +248,7 @@ void testUploadCallable_startFailure_failsFuture() { .thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("start failed"))); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + callable.futureCall("resource-path", streamOf("data")); ExecutionException exception = assertThrows(ExecutionException.class, future::get); assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); @@ -263,7 +263,7 @@ void testUploadCallable_chunkFailure_failsFuture() { .thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("chunk error"))); ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + callable.futureCall("resource-path", streamOf("data")); ExecutionException exception = assertThrows(ExecutionException.class, future::get); assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); @@ -277,7 +277,7 @@ void testUploadCallable_closesPayloadOnSuccess() throws Exception { .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done"))); TrackableStream stream = new TrackableStream("data"); - callable.futureCall("resource-path", stream, null).get(); + callable.futureCall("resource-path", stream).get(); assertThat(stream.closed).isTrue(); } @@ -288,7 +288,7 @@ void testUploadCallable_closesPayloadOnFailure() { .thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("start failed"))); TrackableStream stream = new TrackableStream("data"); - ResumableUploadFuture future = callable.futureCall("resource-path", stream, null); + ResumableUploadFuture future = callable.futureCall("resource-path", stream); assertThrows(ExecutionException.class, future::get); assertThat(stream.closed).isTrue(); @@ -306,7 +306,7 @@ void testUploadCallable_closesPayloadOnCancel() throws Exception { }); TrackableStream stream = new TrackableStream("data"); - ResumableUploadFuture future = callable.futureCall("resource-path", stream, null); + ResumableUploadFuture future = callable.futureCall("resource-path", stream); assertThat(chunkStarted.await(5, TimeUnit.SECONDS)).isTrue(); future.cancel(true); @@ -319,7 +319,7 @@ void testUploadCallable_closesPayloadOnStartSyncFailure() { .thenThrow(new RuntimeException("sync start failure")); TrackableStream stream = new TrackableStream("data"); - ResumableUploadFuture future = callable.futureCall("resource-path", stream, null); + ResumableUploadFuture future = callable.futureCall("resource-path", stream); ExecutionException exception = assertThrows(ExecutionException.class, future::get); assertThat(exception.getCause()).isInstanceOf(RuntimeException.class); @@ -348,7 +348,7 @@ public void execute(Runnable command) { ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "executor-done"))); ResumableUploadFuture future = - customCallable.futureCall("resource-path", streamOf("data"), null); + customCallable.futureCall("resource-path", streamOf("data")); assertThat(future.get()).isEqualTo("executor-done"); assertThat(tasksRun.get()).isGreaterThan(0); } finally { @@ -357,6 +357,98 @@ public void execute(Runnable command) { } } + @Test + void testUploadCallable_convenienceOverload_noContext() throws Exception { + stubStartSession("https://upload.url/convenience"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-conv"))); + + ResumableUploadFuture future = callable.futureCall("resource-path", streamOf("data")); + assertThat(future.get()).isEqualTo("done-conv"); + } + + @Test + void testUploadCallable_withApiCallContext_mergesAndPassesContext() throws Exception { + stubStartSession("https://upload.url/context"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-ctx"))); + + ApiCallContext customContext = + FakeCallContext.createDefault() + .withExtraHeaders( + java.util.Collections.singletonMap( + "X-Custom", java.util.Collections.singletonList("val"))); + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("data"), customContext); + assertThat(future.get()).isEqualTo("done-ctx"); + + ArgumentCaptor startContextCaptor = + ArgumentCaptor.forClass(ApiCallContext.class); + verify(mockStartCallable).futureCall(any(), startContextCaptor.capture()); + assertThat(startContextCaptor.getValue()).isNotNull(); + assertThat(((FakeCallContext) startContextCaptor.getValue()).getExtraHeaders()) + .containsKey("X-Custom"); + + ArgumentCaptor chunkContextCaptor = + ArgumentCaptor.forClass(ApiCallContext.class); + verify(mockChunkCallable).futureCall(any(), chunkContextCaptor.capture()); + assertThat(chunkContextCaptor.getValue()).isNotNull(); + assertThat(((FakeCallContext) chunkContextCaptor.getValue()).getExtraHeaders()) + .doesNotContainKey("X-Custom"); + } + + @Test + void testUploadCallable_withSettings_mergesAndAppliesSettings() throws Exception { + stubStartSession("https://upload.url/settings"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-settings"))); + + ResumableUploadCallSettings customSettings = + ResumableUploadCallSettings.newBuilder().setChunkSize(16).build(); + + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("data"), null, customSettings); + assertThat(future.get()).isEqualTo("done-settings"); + } + + @Test + void testUploadCallable_withSettingsConvenienceOverload_mergesAndAppliesSettings() + throws Exception { + stubStartSession("https://upload.url/settings-convenience"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn( + ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-settings-conv"))); + + ResumableUploadCallSettings customSettings = + ResumableUploadCallSettings.newBuilder().setChunkSize(16).build(); + + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("data"), customSettings); + assertThat(future.get()).isEqualTo("done-settings-conv"); + } + + @Test + void testUploadCallable_withContextAndSettings_appliesBoth() throws Exception { + stubStartSession("https://upload.url/ctx-settings"); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-both"))); + + FakeCallContext customContext = FakeCallContext.createDefault(); + ResumableUploadCallSettings customSettings = + ResumableUploadCallSettings.newBuilder().setChunkSize(16).build(); + + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("data"), customContext, customSettings); + assertThat(future.get()).isEqualTo("done-both"); + } + + @Test + void testResumeCall_throwsUnsupportedOperationException() { + assertThrows( + UnsupportedOperationException.class, + () -> callable.resumeCall("https://upload.url/session", streamOf("data"), null)); + } + private void stubStartSession(String uploadUrl) { when(mockStartCallable.futureCall(any(), any())) .thenReturn(