Skip to content

Commit 1463b4d

Browse files
whoweswhowes
committed
feat(gax): add ApiCallContext and request-level settings overloads to ResumableUploadCallable
Add 2-, 3-, and 4-argument overloads to ResumableUploadCallable for futureCall and resumeCall, supporting: 1. Default context and settings (used by generated ServiceClient convenience methods). 2. Per-request ApiCallContext overrides for transport metadata (extra headers, credentials). 3. Per-request ResumableUploadCallSettings overrides for state-machine knobs. 4. Full method accepting both call context and settings overrides. ResumableUploadCallableImpl performs a 3-tier precedence merge on settings and merges ApiCallContext.
1 parent 7253511 commit 1463b4d

3 files changed

Lines changed: 171 additions & 24 deletions

File tree

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

Lines changed: 59 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -48,19 +48,75 @@ public abstract class ResumableUploadCallable<RequestT, ResponseT> {
4848
protected ResumableUploadCallable() {}
4949

5050
/**
51-
* Performs a new resumable upload asynchronously.
51+
* Performs a new resumable upload asynchronously with default call context and default settings.
5252
*
5353
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
5454
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
5555
* cancellation.
5656
*
5757
* @param request the request message
5858
* @param payload the data payload input stream to upload and close
59-
* @param settings call settings overrides; may be {@code null}
59+
* @return future for tracking and controlling the upload
60+
*/
61+
public ResumableUploadFuture<ResponseT> futureCall(RequestT request, InputStream payload) {
62+
return futureCall(request, payload, null, null);
63+
}
64+
65+
/**
66+
* Performs a new resumable upload asynchronously with a call context override and default
67+
* settings.
68+
*
69+
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
70+
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
71+
* cancellation.
72+
*
73+
* @param request the request message
74+
* @param payload the data payload input stream to upload and close
75+
* @param context call context overrides (e.g. extra headers, credentials, timeout); may be {@code
76+
* null}
77+
* @return future for tracking and controlling the upload
78+
*/
79+
public ResumableUploadFuture<ResponseT> futureCall(
80+
RequestT request, InputStream payload, @Nullable ApiCallContext context) {
81+
return futureCall(request, payload, context, null);
82+
}
83+
84+
/**
85+
* Performs a new resumable upload asynchronously with settings overrides and default call
86+
* context.
87+
*
88+
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
89+
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
90+
* cancellation.
91+
*
92+
* @param request the request message
93+
* @param payload the data payload input stream to upload and close
94+
* @param settings request-level call settings overrides; may be {@code null}
95+
* @return future for tracking and controlling the upload
96+
*/
97+
public ResumableUploadFuture<ResponseT> futureCall(
98+
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
99+
return futureCall(request, payload, null, settings);
100+
}
101+
102+
/**
103+
* Performs a new resumable upload asynchronously with call context and settings overrides.
104+
*
105+
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
106+
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
107+
* cancellation.
108+
*
109+
* @param request the request message
110+
* @param payload the data payload input stream to upload and close
111+
* @param context call context overrides; may be {@code null}
112+
* @param settings request-level call settings overrides; may be {@code null}
60113
* @return future for tracking and controlling the upload
61114
*/
62115
public abstract ResumableUploadFuture<ResponseT> futureCall(
63-
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings);
116+
RequestT request,
117+
InputStream payload,
118+
@Nullable ApiCallContext context,
119+
@Nullable ResumableUploadCallSettings settings);
64120

65121
/**
66122
* Resumes an existing resumable upload session asynchronously using a saved session URL.

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -75,14 +75,18 @@ public ResumableUploadCallableImpl(
7575

7676
@Override
7777
public ResumableUploadFuture<ResponseT> futureCall(
78-
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
78+
RequestT request,
79+
InputStream payload,
80+
@Nullable ApiCallContext context,
81+
@Nullable ResumableUploadCallSettings settings) {
7982
checkNotNull(request, "request must not be null");
8083
checkNotNull(payload, "payload must not be null");
8184
ResumableUploadCallSettings effectiveSettings = defaultCallSettings.merge(settings);
85+
ApiCallContext effectiveCallContext = defaultCallContext.merge(context);
8286

8387
ApiFuture<ResumableUploadSession> startFuture;
8488
try {
85-
startFuture = client.startUploadCallable().futureCall(request, defaultCallContext);
89+
startFuture = client.startUploadCallable().futureCall(request, effectiveCallContext);
8690
} catch (Throwable t) {
8791
startFuture = ApiFutures.immediateFailedFuture(t);
8892
}

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

Lines changed: 106 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -111,8 +111,7 @@ void testUploadCallable_singleChunk_happyPath() throws Exception {
111111
.thenReturn(
112112
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-single")));
113113

114-
ResumableUploadFuture<String> future =
115-
callable.futureCall("resource-path", streamOf("hello"), null);
114+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", streamOf("hello"));
116115

117116
assertThat(future.get()).isEqualTo("response-single");
118117
assertThat(future.isDone()).isTrue();
@@ -136,7 +135,7 @@ void testUploadCallable_multiChunk_happyPath() throws Exception {
136135
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-multi")));
137136

138137
ResumableUploadFuture<String> future =
139-
callable.futureCall("resource-path", streamOf("01234567890123456789"), null);
138+
callable.futureCall("resource-path", streamOf("01234567890123456789"));
140139

141140
assertThat(future.get()).isEqualTo("response-multi");
142141
assertThat(future.isDone()).isTrue();
@@ -156,7 +155,7 @@ void testUploadCallable_zeroByteUpload_finalizesSuccessfully() throws Exception
156155
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-zero")));
157156

158157
ResumableUploadFuture<String> future =
159-
callable.futureCall("resource-path", new ByteArrayInputStream(new byte[0]), null);
158+
callable.futureCall("resource-path", new ByteArrayInputStream(new byte[0]));
160159

161160
assertThat(future.get()).isEqualTo("response-zero");
162161

@@ -176,7 +175,7 @@ void testUploadCallable_singleChunkWithSeparateZeroByteFinalize_completesSuccess
176175
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "response-exact-single")));
177176

178177
ResumableUploadFuture<String> future =
179-
callable.futureCall("resource-path", streamOf("12345678"), null);
178+
callable.futureCall("resource-path", streamOf("12345678"));
180179

181180
assertThat(future.get()).isEqualTo("response-exact-single");
182181

@@ -193,8 +192,7 @@ void testUploadCallable_nullResponse_completesSuccessfully() throws Exception {
193192
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
194193
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, null)));
195194

196-
ResumableUploadFuture<String> future =
197-
callable.futureCall("resource-path", streamOf("data"), null);
195+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", streamOf("data"));
198196

199197
assertThat(future.get()).isNull();
200198
assertThat(future.isDone()).isTrue();
@@ -217,7 +215,7 @@ void testUploadCallable_cancelInFlight_haltsUpload() throws Exception {
217215
});
218216

219217
ResumableUploadFuture<String> future =
220-
callable.futureCall("resource-path", streamOf("0123456789ABCDEF"), null);
218+
callable.futureCall("resource-path", streamOf("0123456789ABCDEF"));
221219

222220
assertThat(chunkStarted.await(5, TimeUnit.SECONDS)).isTrue();
223221
assertThat(future.cancel(true)).isTrue();
@@ -232,8 +230,7 @@ void testUploadCallable_cancelInFlight_haltsUpload() throws Exception {
232230
void testUploadCallable_setInFlightFutureAfterCancel_immediatelyCancelsFuture() {
233231
SettableApiFuture<ResumableUploadSession> startFuture = SettableApiFuture.create();
234232
when(mockStartCallable.futureCall(any(), any())).thenReturn(startFuture);
235-
ResumableUploadFuture<String> future =
236-
callable.futureCall("resource-path", streamOf("data"), null);
233+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", streamOf("data"));
237234
assertThat(future.cancel(true)).isTrue();
238235
assertThat(future.isCancelled()).isTrue();
239236

@@ -247,8 +244,7 @@ void testUploadCallable_startFailure_failsFuture() {
247244
when(mockStartCallable.futureCall(any(), any()))
248245
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("start failed")));
249246

250-
ResumableUploadFuture<String> future =
251-
callable.futureCall("resource-path", streamOf("data"), null);
247+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", streamOf("data"));
252248

253249
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
254250
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
@@ -262,8 +258,7 @@ void testUploadCallable_chunkFailure_failsFuture() {
262258
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
263259
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("chunk error")));
264260

265-
ResumableUploadFuture<String> future =
266-
callable.futureCall("resource-path", streamOf("data"), null);
261+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", streamOf("data"));
267262

268263
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
269264
assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
@@ -277,7 +272,7 @@ void testUploadCallable_closesPayloadOnSuccess() throws Exception {
277272
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done")));
278273

279274
TrackableStream stream = new TrackableStream("data");
280-
callable.futureCall("resource-path", stream, null).get();
275+
callable.futureCall("resource-path", stream).get();
281276

282277
assertThat(stream.closed).isTrue();
283278
}
@@ -288,7 +283,7 @@ void testUploadCallable_closesPayloadOnFailure() {
288283
.thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("start failed")));
289284

290285
TrackableStream stream = new TrackableStream("data");
291-
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream, null);
286+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream);
292287
assertThrows(ExecutionException.class, future::get);
293288

294289
assertThat(stream.closed).isTrue();
@@ -306,7 +301,7 @@ void testUploadCallable_closesPayloadOnCancel() throws Exception {
306301
});
307302

308303
TrackableStream stream = new TrackableStream("data");
309-
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream, null);
304+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream);
310305
assertThat(chunkStarted.await(5, TimeUnit.SECONDS)).isTrue();
311306
future.cancel(true);
312307

@@ -319,7 +314,7 @@ void testUploadCallable_closesPayloadOnStartSyncFailure() {
319314
.thenThrow(new RuntimeException("sync start failure"));
320315

321316
TrackableStream stream = new TrackableStream("data");
322-
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream, null);
317+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", stream);
323318

324319
ExecutionException exception = assertThrows(ExecutionException.class, future::get);
325320
assertThat(exception.getCause()).isInstanceOf(RuntimeException.class);
@@ -348,7 +343,7 @@ public void execute(Runnable command) {
348343
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "executor-done")));
349344

350345
ResumableUploadFuture<String> future =
351-
customCallable.futureCall("resource-path", streamOf("data"), null);
346+
customCallable.futureCall("resource-path", streamOf("data"));
352347
assertThat(future.get()).isEqualTo("executor-done");
353348
assertThat(tasksRun.get()).isGreaterThan(0);
354349
} finally {
@@ -357,6 +352,98 @@ public void execute(Runnable command) {
357352
}
358353
}
359354

355+
@Test
356+
void testUploadCallable_convenienceOverload_noContext() throws Exception {
357+
stubStartSession("https://upload.url/convenience");
358+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
359+
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-conv")));
360+
361+
ResumableUploadFuture<String> future = callable.futureCall("resource-path", streamOf("data"));
362+
assertThat(future.get()).isEqualTo("done-conv");
363+
}
364+
365+
@Test
366+
void testUploadCallable_withApiCallContext_mergesAndPassesContext() throws Exception {
367+
stubStartSession("https://upload.url/context");
368+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
369+
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-ctx")));
370+
371+
ApiCallContext customContext =
372+
FakeCallContext.createDefault()
373+
.withExtraHeaders(
374+
java.util.Collections.singletonMap(
375+
"X-Custom", java.util.Collections.singletonList("val")));
376+
ResumableUploadFuture<String> future =
377+
callable.futureCall("resource-path", streamOf("data"), customContext);
378+
assertThat(future.get()).isEqualTo("done-ctx");
379+
380+
ArgumentCaptor<ApiCallContext> startContextCaptor =
381+
ArgumentCaptor.forClass(ApiCallContext.class);
382+
verify(mockStartCallable).futureCall(any(), startContextCaptor.capture());
383+
assertThat(startContextCaptor.getValue()).isNotNull();
384+
assertThat(((FakeCallContext) startContextCaptor.getValue()).getExtraHeaders())
385+
.containsKey("X-Custom");
386+
387+
ArgumentCaptor<ApiCallContext> chunkContextCaptor =
388+
ArgumentCaptor.forClass(ApiCallContext.class);
389+
verify(mockChunkCallable).futureCall(any(), chunkContextCaptor.capture());
390+
assertThat(chunkContextCaptor.getValue()).isNotNull();
391+
assertThat(((FakeCallContext) chunkContextCaptor.getValue()).getExtraHeaders())
392+
.doesNotContainKey("X-Custom");
393+
}
394+
395+
@Test
396+
void testUploadCallable_withSettings_mergesAndAppliesSettings() throws Exception {
397+
stubStartSession("https://upload.url/settings");
398+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
399+
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-settings")));
400+
401+
ResumableUploadCallSettings customSettings =
402+
ResumableUploadCallSettings.newBuilder().setChunkSize(16).build();
403+
404+
ResumableUploadFuture<String> future =
405+
callable.futureCall("resource-path", streamOf("data"), null, customSettings);
406+
assertThat(future.get()).isEqualTo("done-settings");
407+
}
408+
409+
@Test
410+
void testUploadCallable_withSettingsConvenienceOverload_mergesAndAppliesSettings()
411+
throws Exception {
412+
stubStartSession("https://upload.url/settings-convenience");
413+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
414+
.thenReturn(
415+
ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-settings-conv")));
416+
417+
ResumableUploadCallSettings customSettings =
418+
ResumableUploadCallSettings.newBuilder().setChunkSize(16).build();
419+
420+
ResumableUploadFuture<String> future =
421+
callable.futureCall("resource-path", streamOf("data"), customSettings);
422+
assertThat(future.get()).isEqualTo("done-settings-conv");
423+
}
424+
425+
@Test
426+
void testUploadCallable_withContextAndSettings_appliesBoth() throws Exception {
427+
stubStartSession("https://upload.url/ctx-settings");
428+
when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
429+
.thenReturn(ApiFutures.immediateFuture(ChunkUploadResponse.create(true, "done-both")));
430+
431+
FakeCallContext customContext = FakeCallContext.createDefault();
432+
ResumableUploadCallSettings customSettings =
433+
ResumableUploadCallSettings.newBuilder().setChunkSize(16).build();
434+
435+
ResumableUploadFuture<String> future =
436+
callable.futureCall("resource-path", streamOf("data"), customContext, customSettings);
437+
assertThat(future.get()).isEqualTo("done-both");
438+
}
439+
440+
@Test
441+
void testResumeCall_throwsUnsupportedOperationException() {
442+
assertThrows(
443+
UnsupportedOperationException.class,
444+
() -> callable.resumeCall("https://upload.url/session", streamOf("data"), null));
445+
}
446+
360447
private void stubStartSession(String uploadUrl) {
361448
when(mockStartCallable.futureCall(any(), any()))
362449
.thenReturn(

0 commit comments

Comments
 (0)