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
210 changes: 197 additions & 13 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 @@ -33,6 +33,8 @@
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 All @@ -47,6 +49,8 @@
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.ByteSource;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
Expand All @@ -67,14 +71,20 @@ public final class HttpJsonResumableUploadClient implements ResumableUploadClien

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 +109,37 @@ public String getPath(StartUploadRequest request) {

@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 ResumableUploadRequestFormatter<ChunkUploadRequest>() {
@Override
public Map<String, List<String>> getQueryParamNames(ChunkUploadRequest request) {
return Collections.emptyMap();
}

@Override
public ByteSource getBinaryRequestBody(ChunkUploadRequest request) {
return request.getPayload();
}
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 +181,51 @@ public ApiFuture<ResumableUploadSession> futureCall(
};
}

@Override
public UnaryCallable<ChunkUploadRequest, ChunkUploadResponse> uploadChunkCallable() {
return new UnaryCallable<ChunkUploadRequest, ChunkUploadResponse>() {
@Override
public ApiFuture<ChunkUploadResponse> futureCall(
ChunkUploadRequest request, @Nullable ApiCallContext inputContext) {
Preconditions.checkNotNull(request);
boolean isPayloadEmpty;
try {
isPayloadEmpty = request.getPayload().isEmpty();
} catch (IOException e) {
isPayloadEmpty = false;
}
String command;
if (request.isFinal()) {
command = !isPayloadEmpty ? "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)
.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,25 +275,124 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
/* 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) {
} catch (Exception e) {
future.setException(
ApiExceptionFactory.createException(
"Internal error processing start upload response",
t,
e,
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
/* retryable= */ false));
}
}
}

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 (Exception e) {
future.setException(
ApiExceptionFactory.createException(
"Internal error processing upload chunk response",
e,
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 (isUploadFinal(trailers)) {
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