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 31f15ef83ff7..318cdf308d51 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 @@ -70,15 +70,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 = clientContext.getDefaultCallContext().merge(context); ApiFuture startFuture; try { - startFuture = - client.startUploadCallable().futureCall(request, clientContext.getDefaultCallContext()); + 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 82194bb060fc..c4d8dabbe926 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 @@ -97,8 +97,7 @@ void testUploadCallable_singleChunk_happyPath() throws Exception { .thenReturn( ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-single"))); - ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("hello"), null); + ResumableUploadFuture future = callable.futureCall("resource-path", streamOf("hello")); assertThat(future.get()).isEqualTo("response-single"); assertThat(future.isDone()).isTrue(); @@ -122,7 +121,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(); @@ -142,7 +141,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"); @@ -162,7 +161,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"); @@ -179,8 +178,7 @@ void testUploadCallable_nullResponse_completesSuccessfully() throws Exception { when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) .thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, null))); - ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + ResumableUploadFuture future = callable.futureCall("resource-path", streamOf("data")); assertThat(future.get()).isNull(); assertThat(future.isDone()).isTrue(); @@ -203,7 +201,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(); @@ -218,8 +216,7 @@ void testUploadCallable_cancelInFlight_haltsUpload() throws Exception { void testUploadCallable_setInFlightFutureAfterCancel_immediatelyCancelsFuture() { SettableApiFuture startFuture = SettableApiFuture.create(); when(mockStartCallable.futureCall(any(), any())).thenReturn(startFuture); - ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + ResumableUploadFuture future = callable.futureCall("resource-path", streamOf("data")); assertThat(future.cancel(true)).isTrue(); assertThat(future.isCancelled()).isTrue(); @@ -233,8 +230,7 @@ void testUploadCallable_startFailure_failsFuture() { when(mockStartCallable.futureCall(any(), any())) .thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("start failed"))); - ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + ResumableUploadFuture future = callable.futureCall("resource-path", streamOf("data")); ExecutionException exception = assertThrows(ExecutionException.class, future::get); assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); @@ -248,8 +244,7 @@ void testUploadCallable_chunkFailure_failsFuture() { when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) .thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("chunk error"))); - ResumableUploadFuture future = - callable.futureCall("resource-path", streamOf("data"), null); + ResumableUploadFuture future = callable.futureCall("resource-path", streamOf("data")); ExecutionException exception = assertThrows(ExecutionException.class, future::get); assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); @@ -263,7 +258,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(); } @@ -274,7 +269,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(); @@ -292,7 +287,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); @@ -305,7 +300,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); @@ -313,6 +308,98 @@ void testUploadCallable_closesPayloadOnStartSyncFailure() { assertThat(stream.closed).isTrue(); } + @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(