Skip to content

Commit 1a0efb3

Browse files
whoweswhowes
committed
feat(gax): wire-level changes for resuming uploads
1 parent 6c46294 commit 1a0efb3

7 files changed

Lines changed: 158 additions & 53 deletions

File tree

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

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -29,13 +29,17 @@
2929
*/
3030
package com.google.api.gax.httpjson;
3131

32+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER;
33+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_OFFSET_HEADER;
34+
3235
import com.google.api.client.http.HttpMethods;
3336
import com.google.api.core.ApiFuture;
3437
import com.google.api.gax.resumable.ChunkUploadRequest;
3538
import com.google.api.gax.resumable.ChunkUploadResponse;
3639
import com.google.api.gax.resumable.ResumableUploadStatus;
3740
import com.google.api.gax.rpc.ApiCallContext;
3841
import com.google.api.gax.rpc.ClientContext;
42+
import com.google.api.gax.rpc.EndpointContext;
3943
import com.google.api.gax.rpc.UnaryCallable;
4044
import com.google.api.pathtemplate.PathTemplate;
4145
import com.google.common.base.Preconditions;
@@ -55,10 +59,6 @@
5559
class ResumableUploadChunkCallable<ResponseT>
5660
extends UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> {
5761

58-
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
59-
private static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset";
60-
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
61-
6262
private static final String COMMAND_UPLOAD = "upload";
6363
private static final String COMMAND_FINALIZE = "finalize";
6464
private static final String COMMAND_UPLOAD_FINALIZE = "upload, finalize";
@@ -129,6 +129,9 @@ public ApiFuture<ChunkUploadResponse<ResponseT>> futureCall(
129129
HttpJsonCallContext.createDefault()
130130
.nullToSelf(clientContext.getDefaultCallContext())
131131
.merge(inputContext)
132+
// Session URLs are self-authenticating; omit credentials and reset EndpointContext.
133+
.withCredentials(null)
134+
.withEndpointContext(EndpointContext.getDefaultInstance())
132135
.withExtraHeaders(chunkHeaders);
133136

134137
HttpJsonClientCall<ChunkUploadRequest, String> clientCall =
@@ -175,10 +178,7 @@ private ChunkUploadResponseListener(
175178

176179
@Override
177180
public void onHeaders(HttpJsonMetadata responseHeaders) {
178-
Map<String, Object> headers = responseHeaders.getHeaders();
179-
this.uploadStatus =
180-
ResumableUploadStatus.fromHeader(
181-
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
181+
this.uploadStatus = ResumableUploadHeaders.parseUploadStatus(responseHeaders);
182182
}
183183

184184
@Override

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

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Redistribution and use in source and binary forms, with or without
5+
* modification, are permitted provided that the following conditions are
6+
* met:
7+
*
8+
* * Redistributions of source code must retain the above copyright
9+
* notice, this list of conditions and the following disclaimer.
10+
* * Redistributions in binary form must reproduce the above
11+
* copyright notice, this list of conditions and the following disclaimer
12+
* in the documentation and/or other materials provided with the
13+
* distribution.
14+
* * Neither the name of Google LLC nor the names of its
15+
* contributors may be used to endorse or promote products derived from
16+
* this software without specific prior written permission.
17+
*
18+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
21+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
24+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
25+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
26+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
28+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29+
*/
30+
package com.google.api.gax.httpjson;
31+
32+
import com.google.api.gax.resumable.ResumableUploadSession;
33+
import com.google.api.gax.resumable.ResumableUploadStatus;
34+
import com.google.api.gax.rpc.ApiExceptionFactory;
35+
import com.google.api.gax.rpc.StatusCode;
36+
import com.google.common.base.Strings;
37+
import org.jspecify.annotations.NullMarked;
38+
39+
/** Shared header constants and parsing utilities for HTTP/JSON resumable upload wire calls. */
40+
@NullMarked
41+
final class ResumableUploadHeaders {
42+
43+
static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol";
44+
static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
45+
static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL";
46+
static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity";
47+
static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset";
48+
static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
49+
static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received";
50+
51+
private ResumableUploadHeaders() {}
52+
53+
static ResumableUploadStatus parseUploadStatus(HttpJsonMetadata responseHeaders) {
54+
return ResumableUploadStatus.fromHeader(
55+
HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_STATUS_HEADER));
56+
}
57+
58+
static int parseChunkGranularity(HttpJsonMetadata responseHeaders) {
59+
String granularityStr =
60+
HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_GRANULARITY_HEADER);
61+
if (Strings.isNullOrEmpty(granularityStr)) {
62+
return ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY;
63+
}
64+
try {
65+
int parsed = Integer.parseInt(granularityStr);
66+
if (parsed <= 0) {
67+
throw ApiExceptionFactory.createException(
68+
"Response contained non-positive "
69+
+ UPLOAD_GRANULARITY_HEADER
70+
+ " header: "
71+
+ granularityStr,
72+
/* cause= */ null,
73+
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
74+
/* retryable= */ false);
75+
}
76+
return parsed;
77+
} catch (NumberFormatException e) {
78+
throw ApiExceptionFactory.createException(
79+
"Response contained invalid " + UPLOAD_GRANULARITY_HEADER + " header: " + granularityStr,
80+
e,
81+
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
82+
/* retryable= */ false);
83+
}
84+
}
85+
}

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

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -29,14 +29,19 @@
2929
*/
3030
package com.google.api.gax.httpjson;
3131

32+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER;
33+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_SIZE_RECEIVED_HEADER;
34+
3235
import com.google.api.client.http.HttpMethods;
3336
import com.google.api.core.ApiFuture;
3437
import com.google.api.gax.resumable.QueryStatusRequest;
3538
import com.google.api.gax.resumable.QueryStatusResponse;
39+
import com.google.api.gax.resumable.ResumableUploadSession;
3640
import com.google.api.gax.resumable.ResumableUploadStatus;
3741
import com.google.api.gax.rpc.ApiCallContext;
3842
import com.google.api.gax.rpc.ApiExceptionFactory;
3943
import com.google.api.gax.rpc.ClientContext;
44+
import com.google.api.gax.rpc.EndpointContext;
4045
import com.google.api.gax.rpc.StatusCode;
4146
import com.google.api.gax.rpc.UnaryCallable;
4247
import com.google.api.pathtemplate.PathTemplate;
@@ -61,9 +66,6 @@
6166
class ResumableUploadQueryStatusCallable<ResponseT>
6267
extends UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> {
6368

64-
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
65-
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
66-
private static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received";
6769
private static final String COMMAND_QUERY = "query";
6870

6971
private static final Map<String, List<String>> QUERY_STATUS_HEADERS =
@@ -119,6 +121,9 @@ public ApiFuture<QueryStatusResponse<ResponseT>> futureCall(
119121
HttpJsonCallContext.createDefault()
120122
.nullToSelf(clientContext.getDefaultCallContext())
121123
.merge(inputContext)
124+
// Session URLs are self-authenticating; omit credentials and reset EndpointContext.
125+
.withCredentials(null)
126+
.withEndpointContext(EndpointContext.getDefaultInstance())
122127
.withExtraHeaders(QUERY_STATUS_HEADERS);
123128

124129
HttpJsonClientCall<QueryStatusRequest, String> clientCall =
@@ -183,6 +188,7 @@ private static class QueryStatusResponseListener<ResponseT>
183188
private final ResumableUploadHttpJsonFuture<QueryStatusResponse<ResponseT>> future;
184189
private final HttpResponseParser<ResponseT> responseParser;
185190
private ResumableUploadStatus uploadStatus = ResumableUploadStatus.UNKNOWN;
191+
private int chunkGranularity = ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY;
186192
@Nullable private Long committedOffset = null;
187193
@Nullable private Throwable headerParsingException;
188194
private String responseBody = "";
@@ -196,12 +202,10 @@ private QueryStatusResponseListener(
196202

197203
@Override
198204
public void onHeaders(HttpJsonMetadata responseHeaders) {
199-
Map<String, Object> headers = responseHeaders.getHeaders();
200-
this.uploadStatus =
201-
ResumableUploadStatus.fromHeader(
202-
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
205+
this.uploadStatus = ResumableUploadHeaders.parseUploadStatus(responseHeaders);
203206
try {
204207
this.committedOffset = parseSizeReceived(responseHeaders);
208+
this.chunkGranularity = ResumableUploadHeaders.parseChunkGranularity(responseHeaders);
205209
} catch (Throwable t) {
206210
this.headerParsingException = t;
207211
}
@@ -229,12 +233,14 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
229233
QueryStatusResponse.<ResponseT>newBuilder()
230234
.setUploadStatus(uploadStatus)
231235
.setResponse(responseParser.parse(stream))
236+
.setChunkGranularity(chunkGranularity)
232237
.build());
233238
} else if (committedOffset != null) {
234239
future.set(
235240
QueryStatusResponse.<ResponseT>newBuilder()
236241
.setCommittedOffset(committedOffset)
237242
.setUploadStatus(uploadStatus)
243+
.setChunkGranularity(chunkGranularity)
238244
.build());
239245
} else {
240246
future.setException(

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

Lines changed: 10 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,10 @@
2929
*/
3030
package com.google.api.gax.httpjson;
3131

32+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER;
33+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_PROTOCOL_HEADER;
34+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_URL_HEADER;
35+
3236
import com.google.api.core.ApiFuture;
3337
import com.google.api.gax.resumable.ResumableUploadSession;
3438
import com.google.api.gax.rpc.ApiCallContext;
@@ -52,11 +56,6 @@
5256
class ResumableUploadStartCallable<RequestT>
5357
extends UnaryCallable<RequestT, ResumableUploadSession> {
5458

55-
private static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol";
56-
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
57-
private static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL";
58-
private static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity";
59-
6059
private static final Map<String, List<String>> START_UPLOAD_HEADERS =
6160
ImmutableMap.of(
6261
UPLOAD_PROTOCOL_HEADER, ImmutableList.of("resumable"),
@@ -108,7 +107,7 @@ static <RequestT> UnaryCallable<RequestT, ResumableUploadSession> create(
108107
/** A listener that parses HTTP response headers to produce a {@link ResumableUploadSession}. */
109108
private static class StartUploadResponseListener extends HttpJsonClientCall.Listener<String> {
110109
private final ResumableUploadHttpJsonFuture<ResumableUploadSession> future;
111-
private int chunkGranularity = 1;
110+
private int chunkGranularity = ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY;
112111
@Nullable private String uploadUrl;
113112
@Nullable private Throwable headerParsingException;
114113

@@ -119,39 +118,16 @@ private StartUploadResponseListener(
119118

120119
@Override
121120
public void onHeaders(HttpJsonMetadata responseHeaders) {
122-
Map<String, Object> headers = responseHeaders.getHeaders();
123-
124-
String url = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_URL_HEADER);
121+
String url =
122+
HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_URL_HEADER);
125123
if (!Strings.isNullOrEmpty(url)) {
126124
this.uploadUrl = url;
127125
}
128126

129-
String granularityStr = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_GRANULARITY_HEADER);
130-
if (Strings.isNullOrEmpty(granularityStr)) {
131-
return;
132-
}
133-
134127
try {
135-
int parsed = Integer.parseInt(granularityStr);
136-
if (parsed <= 0) {
137-
this.headerParsingException =
138-
ApiExceptionFactory.createException(
139-
"Start upload response contained non-positive chunk granularity header: "
140-
+ granularityStr,
141-
/* cause= */ null,
142-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
143-
/* retryable= */ false);
144-
} else {
145-
this.chunkGranularity = parsed;
146-
}
147-
} catch (NumberFormatException e) {
148-
this.headerParsingException =
149-
ApiExceptionFactory.createException(
150-
"Start upload response contained invalid chunk granularity header: "
151-
+ granularityStr,
152-
e,
153-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
154-
/* retryable= */ false);
128+
this.chunkGranularity = ResumableUploadHeaders.parseChunkGranularity(responseHeaders);
129+
} catch (Throwable t) {
130+
this.headerParsingException = t;
155131
}
156132
}
157133

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

Lines changed: 31 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,9 @@
5757
import com.google.api.gax.rpc.StatusCode;
5858
import com.google.api.gax.rpc.UnaryCallSettings;
5959
import com.google.api.pathtemplate.PathTemplate;
60+
import com.google.auth.Credentials;
61+
import com.google.auth.oauth2.AccessToken;
62+
import com.google.auth.oauth2.OAuth2Credentials;
6063
import com.google.common.base.Strings;
6164
import java.io.IOException;
6265
import java.nio.charset.StandardCharsets;
@@ -161,7 +164,7 @@ void startUpload_nonPositiveChunkGranularityHeader_throwsException() {
161164
assertThat(exception.getCause()).isInstanceOf(InternalException.class);
162165
assertThat(exception.getCause())
163166
.hasMessageThat()
164-
.contains("Start upload response contained non-positive chunk granularity header: -256");
167+
.contains("Response contained non-positive X-Goog-Upload-Chunk-Granularity header: -256");
165168
}
166169

167170
@Test
@@ -205,6 +208,7 @@ void startUpload_withPayloadAndQueryParams_sendsCorrectRequest() {
205208
assertThat(transport.capturedHeaders.get("x-goog-upload-protocol"))
206209
.containsExactly("resumable");
207210
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("start");
211+
assertThat(transport.capturedHeaders.get("authorization")).containsExactly("Bearer test-token");
208212
}
209213

210214
@Test
@@ -336,6 +340,7 @@ void uploadChunk_intermediateChunk_sendsUploadCommandAndReturnsActiveStatus() {
336340
assertThat(transport.capturedUrl).isEqualTo(TEST_UPLOAD_URL);
337341
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("upload");
338342
assertThat(transport.capturedHeaders.get("x-goog-upload-offset")).containsExactly("0");
343+
assertThat(transport.capturedHeaders).doesNotContainKey("authorization");
339344
}
340345

341346
@Test
@@ -512,6 +517,25 @@ void queryStatus_activeUpload_returnsCommittedOffset() {
512517
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE);
513518

514519
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("query");
520+
assertThat(transport.capturedHeaders).doesNotContainKey("authorization");
521+
}
522+
523+
@Test
524+
void queryStatus_withChunkGranularity_returnsGranularity() {
525+
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
526+
httpResponse.setStatusCode(200);
527+
httpResponse.addHeader("X-Goog-Upload-Status", "active");
528+
httpResponse.addHeader("X-Goog-Upload-Size-Received", "524288");
529+
httpResponse.addHeader("X-Goog-Upload-Chunk-Granularity", "262144");
530+
531+
HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
532+
QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL);
533+
534+
QueryStatusResponse<String> response = client.queryStatusCallable().call(request);
535+
536+
assertThat(response.getCommittedOffset()).isEqualTo(524288L);
537+
assertThat(response.getChunkGranularity()).isEqualTo(262144);
538+
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE);
515539
}
516540

517541
@Test
@@ -622,10 +646,15 @@ private static HttpJsonResumableUploadClient<TestRequest, String> createClient(
622646
.setHttpTransport(transport)
623647
.build();
624648

649+
Credentials credentials = OAuth2Credentials.create(new AccessToken("test-token", null));
650+
HttpJsonCallContext defaultCallContext =
651+
HttpJsonCallContext.createDefault().withChannel(channel).withCredentials(credentials);
652+
625653
ClientContext clientContext =
626654
ClientContext.newBuilder()
627655
.setTransportChannel(HttpJsonTransportChannel.create(channel))
628-
.setDefaultCallContext(HttpJsonCallContext.createDefault().withChannel(channel))
656+
.setDefaultCallContext(defaultCallContext)
657+
.setCredentials(credentials)
629658
.build();
630659

631660
return HttpJsonResumableUploadClient.create(

‎sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,11 +65,18 @@ public abstract class QueryStatusResponse<ResponseT> {
6565
/** Returns the status of the upload session returned by the server. */
6666
public abstract ResumableUploadStatus getUploadStatus();
6767

68+
/**
69+
* Returns the byte alignment granularity required by the server for chunk uploads, or {@code 1}
70+
* if no alignment restriction was returned.
71+
*/
72+
public abstract int getChunkGranularity();
73+
6874
public abstract Builder<ResponseT> toBuilder();
6975

7076
public static <ResponseT> Builder<ResponseT> newBuilder() {
7177
return new AutoValue_QueryStatusResponse.Builder<ResponseT>()
72-
.setUploadStatus(ResumableUploadStatus.ACTIVE);
78+
.setUploadStatus(ResumableUploadStatus.ACTIVE)
79+
.setChunkGranularity(ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY);
7380
}
7481

7582
@AutoValue.Builder
@@ -80,6 +87,8 @@ public abstract static class Builder<ResponseT> {
8087

8188
public abstract Builder<ResponseT> setUploadStatus(ResumableUploadStatus uploadStatus);
8289

90+
public abstract Builder<ResponseT> setChunkGranularity(int chunkGranularity);
91+
8392
public abstract QueryStatusResponse<ResponseT> build();
8493
}
8594
}

0 commit comments

Comments
 (0)