Skip to content

feat(gax): implement uploadChunkCallable for resumable uploads #14140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
whowes wants to merge 1 commit into from
Draft

feat(gax): implement uploadChunkCallable for resumable uploads #14140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
206 changes: 195 additions & 11 deletions ...gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,15 @@
*/
package com.google.api.gax.httpjson;

import com.google.api.client.http.EmptyContent;
import com.google.api.client.http.HttpContent;
import com.google.api.client.http.HttpMethods;
import com.google.api.client.http.InputStreamContent;
import com.google.api.core.ApiFuture;
import com.google.api.core.InternalApi;
import com.google.api.core.SettableApiFuture;
import com.google.api.gax.resumable.ChunkUploadRequest;
import com.google.api.gax.resumable.ChunkUploadResponse;
import com.google.api.gax.resumable.ResumableUploadClient;
import com.google.api.gax.resumable.ResumableUploadSession;
import com.google.api.gax.resumable.StartUploadRequest;
Expand Down Expand Up @@ -67,14 +72,20 @@

private static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol";
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
private static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset";
private static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL";
private static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity";
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
private static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received";
private static final String STATUS_FINAL = "final";

private static final Map<String, List<String>> START_UPLOAD_HEADERS =
ImmutableMap.of(
UPLOAD_PROTOCOL_HEADER, ImmutableList.of("resumable"),
UPLOAD_COMMAND_HEADER, ImmutableList.of("start"));

private static final PathTemplate PATH_TEMPLATE = PathTemplate.create("{+path}");

private static final ApiMethodDescriptor<StartUploadRequest, String> START_UPLOAD_DESCRIPTOR =
ApiMethodDescriptor.<StartUploadRequest, String>newBuilder()
.setFullMethodName("ResumableUpload/StartUpload")
Expand All @@ -99,7 +110,42 @@

@Override
public PathTemplate getPathTemplate() {
return PathTemplate.create("{+path}");
return PATH_TEMPLATE;
}
})
.setResponseParser(StringHttpResponseParser.create())
.build();

private static final ApiMethodDescriptor<ChunkUploadRequest, String> UPLOAD_CHUNK_DESCRIPTOR =
ApiMethodDescriptor.<ChunkUploadRequest, String>newBuilder()
.setFullMethodName("ResumableUpload/UploadChunk")
.setHttpMethod(HttpMethods.POST)
Comment thread
whowes marked this conversation as resolved.
.setType(ApiMethodDescriptor.MethodType.UNARY)
.setRequestFormatter(
new HttpContentRequestFormatter<ChunkUploadRequest>() {
@Override
public Map<String, List<String>> getQueryParamNames(ChunkUploadRequest request) {
return Collections.emptyMap();
}

@Override
public HttpContent getHttpContent(ChunkUploadRequest request) {
if (!request.getPayload().isEmpty()) {
return new InputStreamContent(
"application/octet-stream", request.getPayload().newInput())
.setLength(request.getPayload().size());
}
return new EmptyContent();
}
Comment thread
whowes marked this conversation as resolved.

@Override
public String getPath(ChunkUploadRequest request) {
return request.getUploadUrl();
}

@Override
public PathTemplate getPathTemplate() {
return PATH_TEMPLATE;
}
})
.setResponseParser(StringHttpResponseParser.create())
Expand Down Expand Up @@ -141,6 +187,45 @@
};
}

@Override
public UnaryCallable<ChunkUploadRequest, ChunkUploadResponse> uploadChunkCallable() {
return new UnaryCallable<ChunkUploadRequest, ChunkUploadResponse>() {
@Override
public ApiFuture<ChunkUploadResponse> futureCall(
ChunkUploadRequest request, @Nullable ApiCallContext inputContext) {
Preconditions.checkNotNull(request);
String command;
if (request.isFinal()) {
command = !request.getPayload().isEmpty() ? "upload, finalize" : "finalize";
} else {
command = "upload";
}
Map<String, List<String>> chunkHeaders =
ImmutableMap.of(
UPLOAD_COMMAND_HEADER,
ImmutableList.of(command),
UPLOAD_OFFSET_HEADER,
ImmutableList.of(String.valueOf(request.getOffset())));

HttpJsonCallContext context =
(HttpJsonCallContext)
HttpJsonCallContext.createDefault()
.nullToSelf(clientContext.getDefaultCallContext())
.merge(inputContext)

Check warning on line 214 in sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java

View check run for this annotation

SonarQubeCloud / [gapic-generator-java-root] SonarCloud Code Analysis

Annotate the parameter with @javax.annotation.Nullable in method 'merge' declaration, or make sure that null can not be passed as argument. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaAcXnAClK7i42CvsrYR&open=AaAcXnAClK7i42CvsrYR&pullRequest=14140
.withExtraHeaders(chunkHeaders);

HttpJsonClientCall<ChunkUploadRequest, String> clientCall =
HttpJsonClientCalls.newCall(UPLOAD_CHUNK_DESCRIPTOR, context);

SettableApiFuture<ChunkUploadResponse> future = SettableApiFuture.create();
HttpJsonClientCalls.startUnaryCall(
clientCall, request, context, new ChunkUploadResponseListener(request, future));

return future;
}
};
}

private static class StartUploadResponseListener extends HttpJsonClientCall.Listener<String> {

private final SettableApiFuture<ResumableUploadSession> future;
Expand Down Expand Up @@ -190,16 +275,7 @@
/* retryable= */ false));
}
} else {
Throwable cause = trailers.getException();
ApiException apiException =
cause != null
? API_EXCEPTION_FACTORY.create(cause)
: ApiExceptionFactory.createException(
"Failed to start upload with status code: " + statusCode,
/* cause= */ null,
HttpJsonStatusCode.of(statusCode),
/* retryable= */ false);
future.setException(apiException);
future.setException(createApiException(statusCode, trailers, "Failed to start upload"));
}
} catch (Throwable t) {
future.setException(
Expand All @@ -211,4 +287,112 @@
}
}
}

private static class ChunkUploadResponseListener extends HttpJsonClientCall.Listener<String> {

private final ChunkUploadRequest request;
private final SettableApiFuture<ChunkUploadResponse> future;
private boolean hasUploadStatusHeader = false;
private boolean isComplete = false;
private long committedOffset = -1L;
private String responseBody = "";

ChunkUploadResponseListener(
ChunkUploadRequest request, SettableApiFuture<ChunkUploadResponse> future) {
this.request = request;
this.future = future;
}

@Override
public void onHeaders(HttpJsonMetadata responseHeaders) {
Map<String, Object> headers = responseHeaders.getHeaders();

String statusStr = HttpHeadersUtils.getFirstHeader(headers, UPLOAD_STATUS_HEADER);
if (statusStr != null) {
this.hasUploadStatusHeader = true;
if (STATUS_FINAL.equalsIgnoreCase(statusStr)) {
this.isComplete = true;
}
}

String sizeReceivedStr =
HttpHeadersUtils.getFirstHeader(headers, UPLOAD_SIZE_RECEIVED_HEADER);
if (!Strings.isNullOrEmpty(sizeReceivedStr)) {

@gemini-code-assist gemini-code-assist Bot Aug 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Using a standard null and empty check avoids any dependency on the Strings class from Guava, ensuring compatibility and avoiding potential compilation issues if the import is missing.

Suggested change
if (!Strings.isNullOrEmpty(sizeReceivedStr)) {
if (sizeReceivedStr != null && !sizeReceivedStr.isEmpty()) {

try {
this.committedOffset = Long.parseLong(sizeReceivedStr);
} catch (NumberFormatException ignored) {
// Ignore invalid/malformed size received header and fall back to local offset
// calculation.
}
}
}

@Override
public void onMessage(@Nullable String message) {
if (message != null) {
this.responseBody = message;
}
}

@Override
public void onClose(int statusCode, HttpJsonMetadata trailers) {
try {
if (statusCode >= 200 && statusCode < 300) {
if (!hasUploadStatusHeader) {
future.setException(
ApiExceptionFactory.createException(
"Upload chunk response did not contain valid X-Goog-Upload-Status header",
/* cause= */ null,
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
/* retryable= */ false));
return;
}
long confirmedOffset =
committedOffset >= 0
? committedOffset
: request.getOffset() + request.getPayload().size();
future.set(
ChunkUploadResponse.create(
confirmedOffset, isComplete, isComplete ? responseBody : ""));
} else {
future.setException(createApiException(statusCode, trailers, "Failed to upload chunk"));
}
} catch (Throwable t) {

Check warning on line 360 in sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java

View check run for this annotation

SonarQubeCloud / [gapic-generator-java-root] SonarCloud Code Analysis

Catch Exception instead of Throwable. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaAcpBbA06s6MWCRGEKK&open=AaAcpBbA06s6MWCRGEKK&pullRequest=14140
future.setException(
ApiExceptionFactory.createException(
"Internal error processing upload chunk response",
t,
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
/* retryable= */ false));
}
}
}

private static boolean isUploadFinal(HttpJsonMetadata responseHeaders) {
String statusStr =
HttpHeadersUtils.getFirstHeader(responseHeaders.getHeaders(), UPLOAD_STATUS_HEADER);
return STATUS_FINAL.equalsIgnoreCase(statusStr);
}

private static ApiException createApiException(
int statusCode, @Nullable HttpJsonMetadata trailers, String actionDescription) {
Throwable cause = trailers != null ? trailers.getException() : null;
if (cause != null) {
ApiException apiException = API_EXCEPTION_FACTORY.create(cause);
if (trailers != null && isUploadFinal(trailers)) {

Check warning on line 382 in sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java

View check run for this annotation

SonarQubeCloud / [gapic-generator-java-root] SonarCloud Code Analysis

Remove this expression which always evaluates to "true" 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaA13je7h8df51YGXcpN&open=AaA13je7h8df51YGXcpN&pullRequest=14140
return ApiExceptionFactory.createException(
apiException.getMessage(),
apiException.getCause(),
apiException.getStatusCode(),
/* retryable= */ false,
apiException.getErrorDetails());
}
return apiException;
}
return ApiExceptionFactory.createException(
actionDescription + " with status code: " + statusCode,
/* cause= */ null,
HttpJsonStatusCode.of(statusCode),
/* retryable= */ false);
}
}
Loading
Loading