diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java new file mode 100644 index 000000000000..298a5752cf57 --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java @@ -0,0 +1,107 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import com.google.api.client.http.HttpMethods; +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.api.gax.resumable.ChunkUploadRequest; +import com.google.api.gax.resumable.ChunkUploadResponse; +import com.google.api.gax.resumable.QueryStatusRequest; +import com.google.api.gax.resumable.QueryStatusResponse; +import com.google.api.gax.resumable.ResumableUploadClient; +import com.google.api.gax.resumable.ResumableUploadSession; +import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.UnaryCallable; +import com.google.common.base.Preconditions; +import org.jspecify.annotations.NullMarked; + +/** + * Implementation of {@link ResumableUploadClient} using HTTP/JSON transport. + * + *

Executes the low-level HTTP wire calls for managing resumable upload sessions. + * + * @param request type for starting an upload + * @param response type of the upload method + */ +@NullMarked +@BetaApi +@InternalApi +public final class HttpJsonResumableUploadClient + implements ResumableUploadClient { + + private final UnaryCallable startUploadCallable; + private final UnaryCallable> + uploadChunkCallable; + private final UnaryCallable> + queryStatusCallable; + + public static HttpJsonResumableUploadClient create( + ClientContext clientContext, ApiMethodDescriptor methodDescriptor) { + return new HttpJsonResumableUploadClient<>(clientContext, methodDescriptor); + } + + private HttpJsonResumableUploadClient( + ClientContext clientContext, ApiMethodDescriptor methodDescriptor) { + Preconditions.checkNotNull(clientContext); + Preconditions.checkNotNull(methodDescriptor); + HttpResponseParser responseParser = + Preconditions.checkNotNull(methodDescriptor.getResponseParser()); + + ApiMethodDescriptor startUploadDescriptor = + ApiMethodDescriptor.newBuilder() + .setFullMethodName(methodDescriptor.getFullMethodName()) + .setHttpMethod(HttpMethods.POST) + .setType(ApiMethodDescriptor.MethodType.UNARY) + .setRequestFormatter(methodDescriptor.getRequestFormatter()) + .setResponseParser(ResumableUploadResponseParser.create()) + .build(); + this.startUploadCallable = + ResumableUploadStartCallable.create(clientContext, startUploadDescriptor); + this.uploadChunkCallable = ResumableUploadChunkCallable.create(clientContext, responseParser); + this.queryStatusCallable = + ResumableUploadQueryStatusCallable.create(clientContext, responseParser); + } + + @Override + public UnaryCallable startUploadCallable() { + return startUploadCallable; + } + + @Override + public UnaryCallable> uploadChunkCallable() { + return uploadChunkCallable; + } + + @Override + public UnaryCallable> queryStatusCallable() { + return queryStatusCallable; + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonStatusRuntimeException.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonStatusRuntimeException.java index af357e0952eb..4c6234bc8a86 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonStatusRuntimeException.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonStatusRuntimeException.java @@ -31,6 +31,7 @@ package com.google.api.gax.httpjson; import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; /** * HTTP status code in RuntimeException form, for propagating status code information via @@ -42,7 +43,8 @@ public class HttpJsonStatusRuntimeException extends RuntimeException { private final int statusCode; - public HttpJsonStatusRuntimeException(int statusCode, String message, Throwable cause) { + public HttpJsonStatusRuntimeException( + int statusCode, @Nullable String message, @Nullable Throwable cause) { super(message, cause); this.statusCode = statusCode; } diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java new file mode 100644 index 000000000000..af0f0a4d11f6 --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java @@ -0,0 +1,228 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import com.google.api.client.http.HttpMethods; +import com.google.api.core.ApiFuture; +import com.google.api.gax.resumable.ChunkUploadRequest; +import com.google.api.gax.resumable.ChunkUploadResponse; +import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ApiExceptionFactory; +import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.StatusCode; +import com.google.api.gax.rpc.UnaryCallable; +import com.google.api.pathtemplate.PathTemplate; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** A {@link UnaryCallable} that transmits individual chunks in a resumable upload session. */ +@NullMarked +class ResumableUploadChunkCallable + extends UnaryCallable> { + + 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_STATUS_HEADER = "X-Goog-Upload-Status"; + private static final String STATUS_FINAL = "final"; + + private static final String COMMAND_UPLOAD = "upload"; + private static final String COMMAND_FINALIZE = "finalize"; + private static final String COMMAND_UPLOAD_FINALIZE = "upload, finalize"; + + private static final PathTemplate PATH_TEMPLATE = PathTemplate.create("**"); + + private static final ApiMethodDescriptor UPLOAD_CHUNK_DESCRIPTOR = + ApiMethodDescriptor.newBuilder() + .setFullMethodName("ResumableUpload/UploadChunk") + .setHttpMethod(HttpMethods.POST) + .setType(ApiMethodDescriptor.MethodType.UNARY) + .setRequestFormatter( + new ResumableUploadChunkRequestFormatter() { + @Override + public Map> getQueryParamNames(ChunkUploadRequest request) { + return Collections.emptyMap(); + } + + @Override + public byte[] getBinaryRequestBody(ChunkUploadRequest request) { + return request.getPayload(); + } + + @Override + public String getPath(ChunkUploadRequest request) { + return request.getUploadUrl(); + } + + @Override + public PathTemplate getPathTemplate() { + return PATH_TEMPLATE; + } + }) + .setResponseParser(ResumableUploadResponseParser.create()) + .build(); + + private final ClientContext clientContext; + private final HttpResponseParser responseParser; + + private ResumableUploadChunkCallable( + ClientContext clientContext, HttpResponseParser responseParser) { + this.clientContext = Preconditions.checkNotNull(clientContext); + this.responseParser = Preconditions.checkNotNull(responseParser); + } + + @Override + public ApiFuture> futureCall( + ChunkUploadRequest request, @Nullable ApiCallContext inputContext) { + Preconditions.checkNotNull(request); + boolean isPayloadEmpty = request.getPayload().length == 0; + String command; + if (request.isFinal()) { + command = !isPayloadEmpty ? COMMAND_UPLOAD_FINALIZE : COMMAND_FINALIZE; + } else { + command = COMMAND_UPLOAD; + } + ImmutableMap.Builder> chunkHeadersBuilder = + ImmutableMap.>builder() + .put(UPLOAD_COMMAND_HEADER, ImmutableList.of(command)); + if (!COMMAND_FINALIZE.equals(command)) { + chunkHeadersBuilder.put( + UPLOAD_OFFSET_HEADER, ImmutableList.of(String.valueOf(request.getOffset()))); + } + Map> chunkHeaders = chunkHeadersBuilder.build(); + + HttpJsonCallContext context = + (HttpJsonCallContext) + HttpJsonCallContext.createDefault() + .nullToSelf(clientContext.getDefaultCallContext()) + .merge(inputContext) + .withExtraHeaders(chunkHeaders); + + HttpJsonClientCall clientCall = + HttpJsonClientCalls.newCall(UPLOAD_CHUNK_DESCRIPTOR, context); + + ResumableUploadHttpJsonFuture> future = + new ResumableUploadHttpJsonFuture<>(clientCall); + HttpJsonClientCalls.startUnaryCall( + clientCall, request, context, new ChunkUploadResponseListener<>(future, responseParser)); + + return future; + } + + static UnaryCallable> create( + ClientContext clientContext, HttpResponseParser responseParser) { + UnaryCallable> rawCallable = + new ResumableUploadChunkCallable<>(clientContext, responseParser); + UnaryCallable> callable = + new HttpJsonExceptionCallable<>( + rawCallable, + // Wire calls do not retry directly; retries are managed by ResumableUploadCallable. + Collections.emptySet()); + return callable.withDefaultCallContext(clientContext.getDefaultCallContext()); + } + + /** + * A listener that processes chunk upload response headers and bodies to produce the {@link + * ChunkUploadResponse}. + */ + private static class ChunkUploadResponseListener + extends HttpJsonClientCall.Listener { + + private final ResumableUploadHttpJsonFuture> future; + private final HttpResponseParser responseParser; + @Nullable private String uploadStatus = null; + private String responseBody = ""; + + private ChunkUploadResponseListener( + ResumableUploadHttpJsonFuture> future, + HttpResponseParser responseParser) { + this.future = future; + this.responseParser = responseParser; + } + + @Override + public void onHeaders(HttpJsonMetadata responseHeaders) { + Map headers = responseHeaders.getHeaders(); + this.uploadStatus = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER); + } + + @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 (uploadStatus == null) { + future.setException( + ApiExceptionFactory.createException( + "Upload chunk response did not contain valid " + + UPLOAD_STATUS_HEADER + + " header", + /* cause= */ null, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false)); + return; + } + boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus); + ChunkUploadResponse.Builder chunkResponseBuilder = + ChunkUploadResponse.newBuilder().setComplete(isComplete); + if (isComplete) { + InputStream stream = + new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8)); + chunkResponseBuilder.setResponse(responseParser.parse(stream)); + } + future.set(chunkResponseBuilder.build()); + } else { + Throwable cause = trailers.getException(); + future.setException( + cause != null + ? cause + : new HttpJsonStatusRuntimeException( + statusCode, "Failed to upload chunk with status code: " + statusCode, null)); + } + } catch (Throwable t) { + future.setException(t); + } + } + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHttpJsonFuture.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHttpJsonFuture.java new file mode 100644 index 000000000000..9221bfd12da8 --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHttpJsonFuture.java @@ -0,0 +1,63 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import com.google.api.core.AbstractApiFuture; +import com.google.api.core.ApiFuture; +import org.jspecify.annotations.NullMarked; + +/** + * An {@link ApiFuture} that cancels the underlying {@link HttpJsonClientCall} upon cancellation to + * prevent connection leaks. + */ +@NullMarked +class ResumableUploadHttpJsonFuture extends AbstractApiFuture { + + private final HttpJsonClientCall call; + + ResumableUploadHttpJsonFuture(HttpJsonClientCall call) { + this.call = call; + } + + @Override + protected void interruptTask() { + call.cancel("Call was cancelled", null); + } + + @Override + public boolean set(T value) { + return super.set(value); + } + + @Override + public boolean setException(Throwable throwable) { + return super.setException(throwable); + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java new file mode 100644 index 000000000000..8176f62bf711 --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java @@ -0,0 +1,262 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import com.google.api.client.http.HttpMethods; +import com.google.api.core.ApiFuture; +import com.google.api.gax.resumable.QueryStatusRequest; +import com.google.api.gax.resumable.QueryStatusResponse; +import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ApiExceptionFactory; +import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.StatusCode; +import com.google.api.gax.rpc.UnaryCallable; +import com.google.api.pathtemplate.PathTemplate; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** + * A {@link UnaryCallable} that queries the committed status and offset of a resumable upload + * session. + */ +@NullMarked +class ResumableUploadQueryStatusCallable + extends UnaryCallable> { + + private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command"; + 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 String COMMAND_QUERY = "query"; + + private static final Map> QUERY_STATUS_HEADERS = + ImmutableMap.of(UPLOAD_COMMAND_HEADER, ImmutableList.of(COMMAND_QUERY)); + + private static final PathTemplate PATH_TEMPLATE = PathTemplate.create("**"); + + private static final ApiMethodDescriptor QUERY_STATUS_DESCRIPTOR = + ApiMethodDescriptor.newBuilder() + .setFullMethodName("ResumableUpload/QueryStatus") + .setHttpMethod(HttpMethods.POST) + .setType(ApiMethodDescriptor.MethodType.UNARY) + .setRequestFormatter( + new HttpRequestFormatter() { + @Override + public Map> getQueryParamNames(QueryStatusRequest request) { + return Collections.emptyMap(); + } + + @Override + public String getRequestBody(QueryStatusRequest request) { + return ""; + } + + @Override + public String getPath(QueryStatusRequest request) { + return request.getUploadUrl(); + } + + @Override + public PathTemplate getPathTemplate() { + return PATH_TEMPLATE; + } + }) + .setResponseParser(ResumableUploadResponseParser.create()) + .build(); + + private final ClientContext clientContext; + private final HttpResponseParser responseParser; + + private ResumableUploadQueryStatusCallable( + ClientContext clientContext, HttpResponseParser responseParser) { + this.clientContext = Preconditions.checkNotNull(clientContext); + this.responseParser = Preconditions.checkNotNull(responseParser); + } + + @Override + public ApiFuture> futureCall( + QueryStatusRequest request, @Nullable ApiCallContext inputContext) { + Preconditions.checkNotNull(request); + HttpJsonCallContext context = + (HttpJsonCallContext) + HttpJsonCallContext.createDefault() + .nullToSelf(clientContext.getDefaultCallContext()) + .merge(inputContext) + .withExtraHeaders(QUERY_STATUS_HEADERS); + + HttpJsonClientCall clientCall = + HttpJsonClientCalls.newCall(QUERY_STATUS_DESCRIPTOR, context); + + ResumableUploadHttpJsonFuture> future = + new ResumableUploadHttpJsonFuture<>(clientCall); + HttpJsonClientCalls.startUnaryCall( + clientCall, request, context, new QueryStatusResponseListener<>(future, responseParser)); + + return future; + } + + static UnaryCallable> create( + ClientContext clientContext, HttpResponseParser responseParser) { + UnaryCallable> rawCallable = + new ResumableUploadQueryStatusCallable<>(clientContext, responseParser); + UnaryCallable> callable = + new HttpJsonExceptionCallable<>( + rawCallable, + // Wire calls do not retry directly; retries are managed by ResumableUploadCallable. + Collections.emptySet()); + return callable.withDefaultCallContext(clientContext.getDefaultCallContext()); + } + + @Nullable + private static Long parseSizeReceived(HttpJsonMetadata responseHeaders) { + String sizeReceivedStr = + HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_SIZE_RECEIVED_HEADER); + if (!Strings.isNullOrEmpty(sizeReceivedStr)) { + try { + long parsed = Long.parseLong(sizeReceivedStr); + if (parsed < 0) { + throw ApiExceptionFactory.createException( + "Response contained negative " + + UPLOAD_SIZE_RECEIVED_HEADER + + " header: " + + sizeReceivedStr, + /* cause= */ null, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false); + } + return parsed; + } catch (NumberFormatException e) { + throw ApiExceptionFactory.createException( + "Response contained invalid " + + UPLOAD_SIZE_RECEIVED_HEADER + + " header: " + + sizeReceivedStr, + e, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false); + } + } + return null; + } + + /** A listener that parses query response headers to produce the {@link QueryStatusResponse}. */ + private static class QueryStatusResponseListener + extends HttpJsonClientCall.Listener { + + private final ResumableUploadHttpJsonFuture> future; + private final HttpResponseParser responseParser; + @Nullable private String uploadStatus = null; + @Nullable private Long committedOffset = null; + @Nullable private Throwable headerParsingException; + private String responseBody = ""; + + private QueryStatusResponseListener( + ResumableUploadHttpJsonFuture> future, + HttpResponseParser responseParser) { + this.future = future; + this.responseParser = responseParser; + } + + @Override + public void onHeaders(HttpJsonMetadata responseHeaders) { + Map headers = responseHeaders.getHeaders(); + this.uploadStatus = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER); + try { + this.committedOffset = parseSizeReceived(responseHeaders); + } catch (Throwable t) { + this.headerParsingException = t; + } + } + + @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 (headerParsingException != null) { + future.setException(headerParsingException); + return; + } + boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus); + if (isComplete) { + QueryStatusResponse.Builder queryResponseBuilder = + QueryStatusResponse.newBuilder().setComplete(true); + InputStream stream = + new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8)); + queryResponseBuilder.setResponse(responseParser.parse(stream)); + future.set(queryResponseBuilder.build()); + } else if (committedOffset != null) { + future.set( + QueryStatusResponse.newBuilder() + .setComplete(false) + .setCommittedOffset(committedOffset) + .build()); + } else { + future.setException( + ApiExceptionFactory.createException( + "Query status response did not contain valid " + + UPLOAD_SIZE_RECEIVED_HEADER + + " header", + /* cause= */ null, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false)); + } + } else { + Throwable cause = trailers.getException(); + future.setException( + cause != null + ? cause + : new HttpJsonStatusRuntimeException( + statusCode, + "Failed to query upload status with status code: " + statusCode, + null)); + } + } catch (Throwable t) { + future.setException(t); + } + } + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java new file mode 100644 index 000000000000..3e546c3e14e6 --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java @@ -0,0 +1,197 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import com.google.api.core.ApiFuture; +import com.google.api.gax.resumable.ResumableUploadSession; +import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ApiExceptionFactory; +import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.StatusCode; +import com.google.api.gax.rpc.UnaryCallable; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** A {@link UnaryCallable} that initiates a resumable upload session. */ +@NullMarked +class ResumableUploadStartCallable + extends UnaryCallable { + + 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_URL_HEADER = "X-Goog-Upload-URL"; + private static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity"; + + private static final Map> START_UPLOAD_HEADERS = + ImmutableMap.of( + UPLOAD_PROTOCOL_HEADER, ImmutableList.of("resumable"), + UPLOAD_COMMAND_HEADER, ImmutableList.of("start")); + + private final ApiMethodDescriptor descriptor; + private final ClientContext clientContext; + + private ResumableUploadStartCallable( + ClientContext clientContext, ApiMethodDescriptor descriptor) { + this.clientContext = Preconditions.checkNotNull(clientContext); + this.descriptor = Preconditions.checkNotNull(descriptor); + } + + @Override + public ApiFuture futureCall( + RequestT request, @Nullable ApiCallContext inputContext) { + Preconditions.checkNotNull(request); + HttpJsonCallContext context = + (HttpJsonCallContext) + HttpJsonCallContext.createDefault() + .nullToSelf(clientContext.getDefaultCallContext()) + .merge(inputContext) + .withExtraHeaders(START_UPLOAD_HEADERS); + + HttpJsonClientCall clientCall = + HttpJsonClientCalls.newCall(descriptor, context); + + ResumableUploadHttpJsonFuture future = + new ResumableUploadHttpJsonFuture<>(clientCall); + HttpJsonClientCalls.startUnaryCall( + clientCall, request, context, new StartUploadResponseListener(future)); + + return future; + } + + static UnaryCallable create( + ClientContext clientContext, ApiMethodDescriptor descriptor) { + UnaryCallable rawCallable = + new ResumableUploadStartCallable<>(clientContext, descriptor); + UnaryCallable callable = + new HttpJsonExceptionCallable<>( + rawCallable, + // Wire calls do not retry directly; retries are managed by ResumableUploadCallable. + Collections.emptySet()); + return callable.withDefaultCallContext(clientContext.getDefaultCallContext()); + } + + /** A listener that parses HTTP response headers to produce a {@link ResumableUploadSession}. */ + private static class StartUploadResponseListener extends HttpJsonClientCall.Listener { + private final ResumableUploadHttpJsonFuture future; + private long chunkGranularity = 1L; + @Nullable private String uploadUrl; + @Nullable private Throwable headerParsingException; + + private StartUploadResponseListener( + ResumableUploadHttpJsonFuture future) { + this.future = future; + } + + @Override + public void onHeaders(HttpJsonMetadata responseHeaders) { + Map headers = responseHeaders.getHeaders(); + + String url = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_URL_HEADER); + if (!Strings.isNullOrEmpty(url)) { + this.uploadUrl = url; + } + + String granularityStr = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_GRANULARITY_HEADER); + if (Strings.isNullOrEmpty(granularityStr)) { + return; + } + + try { + long parsed = Long.parseLong(granularityStr); + if (parsed <= 0) { + this.headerParsingException = + ApiExceptionFactory.createException( + "Start upload response contained non-positive chunk granularity header: " + + granularityStr, + /* cause= */ null, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false); + } else { + this.chunkGranularity = parsed; + } + } catch (NumberFormatException e) { + this.headerParsingException = + ApiExceptionFactory.createException( + "Start upload response contained invalid chunk granularity header: " + + granularityStr, + e, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false); + } + } + + @Override + public void onMessage(@Nullable String message) { + // Response body is not needed for startUpload; session URL is in headers. + } + + @Override + public void onClose(int statusCode, HttpJsonMetadata trailers) { + try { + if (statusCode >= 200 && statusCode < 300) { + if (headerParsingException != null) { + future.setException(headerParsingException); + return; + } + if (!Strings.isNullOrEmpty(uploadUrl)) { + future.set( + ResumableUploadSession.newBuilder() + .setUploadUrl(uploadUrl) + .setChunkGranularity(chunkGranularity) + .build()); + } else { + future.setException( + ApiExceptionFactory.createException( + "Start upload response did not contain upload session URL header", + /* cause= */ null, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false)); + } + } else { + Throwable cause = trailers.getException(); + future.setException( + cause != null + ? cause + : new HttpJsonStatusRuntimeException( + statusCode, "Failed to start upload with status code: " + statusCode, null)); + } + } catch (Throwable t) { + future.setException(t); + } + } + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java new file mode 100644 index 000000000000..2736b3d1c3a8 --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java @@ -0,0 +1,641 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.google.api.client.http.HttpMethods; +import com.google.api.client.http.HttpTransport; +import com.google.api.client.http.LowLevelHttpRequest; +import com.google.api.client.http.LowLevelHttpResponse; +import com.google.api.client.testing.http.MockHttpTransport; +import com.google.api.client.testing.http.MockLowLevelHttpRequest; +import com.google.api.client.testing.http.MockLowLevelHttpResponse; +import com.google.api.core.InternalApi; +import com.google.api.gax.resumable.ChunkUploadRequest; +import com.google.api.gax.resumable.ChunkUploadResponse; +import com.google.api.gax.resumable.QueryStatusRequest; +import com.google.api.gax.resumable.QueryStatusResponse; +import com.google.api.gax.resumable.ResumableUploadSession; +import com.google.api.gax.rpc.AbortedException; +import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ApiException; +import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.InternalException; +import com.google.api.gax.rpc.NotFoundException; +import com.google.api.gax.rpc.StatusCode; +import com.google.api.pathtemplate.PathTemplate; +import com.google.common.base.Strings; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import org.jspecify.annotations.Nullable; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +@InternalApi +class HttpJsonResumableUploadClientTest { + + private static final String TEST_UPLOAD_URL = + "https://test.googleapis.com/upload/session/test-session-id"; + + private static ExecutorService executorService; + + @BeforeAll + static void setUp() { + executorService = Executors.newFixedThreadPool(2); + } + + @AfterAll + static void tearDown() { + executorService.shutdownNow(); + } + + @Test + void startUpload_validHeaders_returnsSession() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-URL", TEST_UPLOAD_URL); + httpResponse.addHeader("X-Goog-Upload-Chunk-Granularity", "262144"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + TestRequest request = new TestRequest("upload/v1/resources"); + + ResumableUploadSession session = client.startUploadCallable().call(request); + + assertThat(session.getUploadUrl()).isEqualTo(TEST_UPLOAD_URL); + assertThat(session.getChunkGranularity()).isEqualTo(262144L); + } + + @Test + void startUpload_caseInsensitiveHeaders_returnsSession() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader( + "x-goog-upload-url", "https://test.googleapis.com/upload/session/case-insensitive"); + httpResponse.addHeader("x-goog-upload-chunk-granularity", "524288"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + TestRequest request = new TestRequest("upload/v1/resources"); + + ResumableUploadSession session = client.startUploadCallable().call(request); + + assertThat(session.getUploadUrl()) + .isEqualTo("https://test.googleapis.com/upload/session/case-insensitive"); + assertThat(session.getChunkGranularity()).isEqualTo(524288L); + } + + @Test + void startUpload_malformedChunkGranularityHeader_throwsException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-URL", TEST_UPLOAD_URL); + httpResponse.addHeader("X-Goog-Upload-Chunk-Granularity", "not-a-number"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + TestRequest request = new TestRequest("upload/v1/resources"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.startUploadCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(InternalException.class); + assertThat(exception.getCause().getCause()).isInstanceOf(NumberFormatException.class); + } + + @Test + void startUpload_nonPositiveChunkGranularityHeader_throwsException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-URL", TEST_UPLOAD_URL); + httpResponse.addHeader("X-Goog-Upload-Chunk-Granularity", "-256"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + TestRequest request = new TestRequest("upload/v1/resources"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.startUploadCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(InternalException.class); + assertThat(exception.getCause()) + .hasMessageThat() + .contains("Start upload response contained non-positive chunk granularity header: -256"); + } + + @Test + void startUpload_missingSessionUrlHeader_throwsException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Chunk-Granularity", "262144"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + TestRequest request = new TestRequest("upload/v1/resources"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.startUploadCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(InternalException.class); + assertThat(exception.getCause()) + .hasMessageThat() + .contains("Start upload response did not contain upload session URL header"); + } + + @Test + void startUpload_withPayloadAndQueryParams_sendsCorrectRequest() { + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + response.addHeader("X-Goog-Upload-URL", TEST_UPLOAD_URL); + + CapturingHttpTransport transport = new CapturingHttpTransport(response); + HttpJsonResumableUploadClient client = createClient(transport); + + Map> queryParams = + Collections.singletonMap("uploadType", Collections.singletonList("resumable")); + TestRequest request = + new TestRequest("upload/v1/resources", "{\"name\":\"my-resource.txt\"}", queryParams); + + client.startUploadCallable().call(request); + + assertThat(transport.capturedUrl) + .isEqualTo("https://test.googleapis.com/upload/v1/resources?uploadType=resumable"); + assertThat(transport.capturedContent).isEqualTo("{\"name\":\"my-resource.txt\"}"); + assertThat(transport.capturedHeaders.get("x-goog-upload-protocol")) + .containsExactly("resumable"); + assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("start"); + } + + @Test + void startUpload_serverReturnsError_throwsApiException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(404); + httpResponse.setContent("{\"error\":{\"message\":\"Resource not found\"}}"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + TestRequest request = new TestRequest("upload/v1/nonexistent"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.startUploadCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(NotFoundException.class); + NotFoundException notFoundException = (NotFoundException) exception.getCause(); + assertThat(notFoundException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.NOT_FOUND); + } + + @Test + void startUpload_withCustomExtraHeaders_preservesHeaders() { + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + response.addHeader("X-Goog-Upload-URL", TEST_UPLOAD_URL); + + CapturingHttpTransport transport = new CapturingHttpTransport(response); + HttpJsonResumableUploadClient client = createClient(transport); + + TestRequest request = new TestRequest("upload/v1/resources"); + Map> customHeaders = + Collections.singletonMap("X-Custom-Header", Collections.singletonList("CustomValue")); + + ApiCallContext callContext = + HttpJsonCallContext.createDefault().withExtraHeaders(customHeaders); + + client.startUploadCallable().call(request, callContext); + + assertThat(transport.capturedHeaders.get("x-custom-header")).containsExactly("CustomValue"); + } + + @Test + void uploadChunk_intermediateChunk_sendsUploadCommandAndReturnsActiveStatus() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "active"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + + byte[] payload = new byte[262144]; + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload(payload) + .setOffset(0L) + .setFinal(false) + .build(); + + ChunkUploadResponse response = client.uploadChunkCallable().call(request); + + assertThat(response.isComplete()).isFalse(); + assertThat(response.getResponse()).isNull(); + + assertThat(transport.capturedUrl).isEqualTo(TEST_UPLOAD_URL); + assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("upload"); + assertThat(transport.capturedHeaders.get("x-goog-upload-offset")).containsExactly("0"); + } + + @Test + void uploadChunk_finalChunk_sendsUploadFinalizeAndReturnsResponseBody() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "final"); + httpResponse.setContent("{\"name\":\"uploaded-file.txt\",\"size\":524288}"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + + byte[] payload = new byte[262144]; + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload(payload) + .setOffset(262144L) + .setFinal(true) + .build(); + + ChunkUploadResponse response = client.uploadChunkCallable().call(request); + + assertThat(response.isComplete()).isTrue(); + assertThat(response.getResponse()) + .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":524288}"); + + assertThat(transport.capturedHeaders.get("x-goog-upload-command")) + .containsExactly("upload, finalize"); + assertThat(transport.capturedHeaders.get("x-goog-upload-offset")).containsExactly("262144"); + } + + @Test + void uploadChunk_emptyPayloadFinal_sendsFinalizeCommandAndReturnsResponseBody() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "final"); + httpResponse.setContent("{\"name\":\"uploaded-file.txt\",\"size\":1048576}"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload(new byte[0]) + .setOffset(1048576L) + .setFinal(true) + .build(); + + ChunkUploadResponse response = client.uploadChunkCallable().call(request); + + assertThat(response.isComplete()).isTrue(); + assertThat(response.getResponse()) + .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}"); + + assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("finalize"); + assertThat(transport.capturedHeaders).doesNotContainKey("x-goog-upload-offset"); + } + + @Test + void uploadChunk_withCustomExtraHeaders_preservesHeaders() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "active"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload("data".getBytes(StandardCharsets.UTF_8)) + .setOffset(0L) + .build(); + + Map> customHeaders = + Collections.singletonMap( + "X-Custom-Chunk-Header", Collections.singletonList("CustomChunkValue")); + + ApiCallContext callContext = + HttpJsonCallContext.createDefault().withExtraHeaders(customHeaders); + + client.uploadChunkCallable().call(request, callContext); + + assertThat(transport.capturedHeaders.get("x-custom-chunk-header")) + .containsExactly("CustomChunkValue"); + } + + @Test + void uploadChunk_serverReturnsConflictOrError_throwsException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(409); + httpResponse.setContent("{\"error\":{\"message\":\"Invalid offset\"}}"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload("data".getBytes(StandardCharsets.UTF_8)) + .setOffset(100L) + .build(); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(AbortedException.class); + AbortedException abortedException = (AbortedException) exception.getCause(); + assertThat(abortedException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.ABORTED); + } + + @Test + void uploadChunk_missingUploadStatusHeader_throwsInternalException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload("data".getBytes(StandardCharsets.UTF_8)) + .setOffset(0L) + .build(); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(InternalException.class); + assertThat(exception.getCause()) + .hasMessageThat() + .contains("Upload chunk response did not contain valid X-Goog-Upload-Status header"); + } + + @Test + void uploadChunk_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(503); + httpResponse.addHeader("X-Goog-Upload-Status", "final"); + httpResponse.setContent("{\"error\":{\"message\":\"Upload rejected by backend\"}}"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload("data".getBytes(StandardCharsets.UTF_8)) + .setOffset(0L) + .build(); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(ApiException.class); + ApiException apiException = (ApiException) exception.getCause(); + assertThat(apiException.isRetryable()).isFalse(); + assertThat(apiException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.UNAVAILABLE); + } + + @Test + void queryStatus_activeUpload_returnsCommittedOffset() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "active"); + httpResponse.addHeader("X-Goog-Upload-Size-Received", "524288"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL); + + QueryStatusResponse response = client.queryStatusCallable().call(request); + + assertThat(response.isComplete()).isFalse(); + assertThat(response.getCommittedOffset()).isEqualTo(524288L); + assertThat(response.getResponse()).isNull(); + + assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("query"); + } + + @Test + void queryStatus_finalUpload_returnsCompleteAndResponseBody() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "final"); + httpResponse.setContent("{\"name\":\"uploaded-file.txt\",\"size\":1048576}"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL); + + QueryStatusResponse response = client.queryStatusCallable().call(request); + + assertThat(response.isComplete()).isTrue(); + assertThat(response.getCommittedOffset()).isNull(); + assertThat(response.getResponse()) + .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}"); + } + + @Test + void queryStatus_withCustomExtraHeaders_preservesHeaders() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "active"); + httpResponse.addHeader("X-Goog-Upload-Size-Received", "256"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL); + + Map> customHeaders = + Collections.singletonMap( + "X-Custom-Query-Header", Collections.singletonList("CustomQueryValue")); + + ApiCallContext callContext = + HttpJsonCallContext.createDefault().withExtraHeaders(customHeaders); + + client.queryStatusCallable().call(request, callContext); + + assertThat(transport.capturedHeaders.get("x-custom-query-header")) + .containsExactly("CustomQueryValue"); + } + + @Test + void queryStatus_serverReturnsError_throwsApiException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(404); + httpResponse.setContent("{\"error\":{\"message\":\"Session not found\"}}"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + QueryStatusRequest request = + QueryStatusRequest.create("https://test.googleapis.com/upload/session/invalid"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.queryStatusCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(NotFoundException.class); + NotFoundException notFoundException = (NotFoundException) exception.getCause(); + assertThat(notFoundException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.NOT_FOUND); + } + + @Test + void queryStatus_missingOrMalformedSizeReceivedHeader_throwsException() { + QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL); + + // Missing header + MockLowLevelHttpResponse missingHeaderResponse = new MockLowLevelHttpResponse(); + missingHeaderResponse.setStatusCode(200); + missingHeaderResponse.addHeader("X-Goog-Upload-Status", "active"); + + HttpJsonResumableUploadClient missingClient = + createClient(missingHeaderResponse); + ExecutionException missingException = + assertThrows( + ExecutionException.class, + () -> missingClient.queryStatusCallable().futureCall(request).get()); + assertThat(missingException.getCause()).isInstanceOf(InternalException.class); + assertThat(missingException.getCause()) + .hasMessageThat() + .contains("Query status response did not contain valid X-Goog-Upload-Size-Received header"); + + // Malformed header + MockLowLevelHttpResponse malformedHeaderResponse = new MockLowLevelHttpResponse(); + malformedHeaderResponse.setStatusCode(200); + malformedHeaderResponse.addHeader("X-Goog-Upload-Status", "active"); + malformedHeaderResponse.addHeader("X-Goog-Upload-Size-Received", "not-a-number"); + + HttpJsonResumableUploadClient malformedClient = + createClient(malformedHeaderResponse); + ExecutionException malformedException = + assertThrows( + ExecutionException.class, + () -> malformedClient.queryStatusCallable().futureCall(request).get()); + assertThat(malformedException.getCause()).isInstanceOf(InternalException.class); + assertThat(malformedException.getCause()) + .hasMessageThat() + .contains("Response contained invalid X-Goog-Upload-Size-Received header: not-a-number"); + } + + private static HttpJsonResumableUploadClient createClient( + HttpTransport transport) { + ManagedHttpJsonChannel channel = + ManagedHttpJsonChannel.newBuilder() + .setEndpoint("test.googleapis.com") + .setExecutor(executorService) + .setHttpTransport(transport) + .build(); + + ClientContext clientContext = + ClientContext.newBuilder() + .setTransportChannel(HttpJsonTransportChannel.create(channel)) + .setDefaultCallContext(HttpJsonCallContext.createDefault().withChannel(channel)) + .build(); + + return HttpJsonResumableUploadClient.create(clientContext, TEST_METHOD_DESCRIPTOR); + } + + private static HttpJsonResumableUploadClient createClient( + MockLowLevelHttpResponse response) { + return createClient(new MockHttpTransport.Builder().setLowLevelHttpResponse(response).build()); + } + + /** A mock transport that captures request URL, headers, and body for verification. */ + private static class CapturingHttpTransport extends MockHttpTransport { + private final MockLowLevelHttpResponse response; + final Map> capturedHeaders = new HashMap<>(); + @Nullable String capturedUrl; + @Nullable String capturedContent; + + CapturingHttpTransport(MockLowLevelHttpResponse response) { + this.response = response; + } + + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + this.capturedUrl = url; + return new MockLowLevelHttpRequest() { + @Override + public LowLevelHttpResponse execute() throws IOException { + capturedHeaders.putAll(getHeaders()); + capturedContent = getContentAsString(); + return response; + } + }; + } + } + + private static final ApiMethodDescriptor TEST_METHOD_DESCRIPTOR = + ApiMethodDescriptor.newBuilder() + .setFullMethodName("ResumableUpload/StartUpload") + .setHttpMethod(HttpMethods.POST) + .setType(ApiMethodDescriptor.MethodType.UNARY) + .setRequestFormatter( + new HttpRequestFormatter() { + @Override + public Map> getQueryParamNames(TestRequest request) { + return request.queryParams; + } + + @Override + public String getRequestBody(TestRequest request) { + return Strings.nullToEmpty(request.jsonPayload); + } + + @Override + public String getPath(TestRequest request) { + return request.path; + } + + @Override + public PathTemplate getPathTemplate() { + return PathTemplate.create("**"); + } + }) + .setResponseParser(ResumableUploadResponseParser.create()) + .build(); + + private static class TestRequest { + final String path; + @Nullable final String jsonPayload; + final Map> queryParams; + + TestRequest(String path) { + this(path, null, Collections.emptyMap()); + } + + TestRequest(String path, @Nullable String jsonPayload, Map> queryParams) { + this.path = path; + this.jsonPayload = jsonPayload; + this.queryParams = queryParams; + } + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadRequest.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadRequest.java new file mode 100644 index 000000000000..73b08d578208 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadRequest.java @@ -0,0 +1,75 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; + +/** Request value object for uploading a chunk to an active resumable upload session. */ +@NullMarked +@BetaApi +@InternalApi +@AutoValue +public abstract class ChunkUploadRequest { + + /** The upload session URL returned during session initialization. */ + public abstract String getUploadUrl(); + + /** The binary chunk payload to upload. */ + @SuppressWarnings("mutable") + public abstract byte[] getPayload(); + + /** The byte offset of this chunk in the overall stream. */ + public abstract long getOffset(); + + /** Whether this is the final chunk in the stream. */ + public abstract boolean isFinal(); + + public abstract Builder toBuilder(); + + public static Builder newBuilder() { + return new AutoValue_ChunkUploadRequest.Builder().setFinal(false); + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setUploadUrl(String uploadUrl); + + public abstract Builder setPayload(byte[] payload); + + public abstract Builder setOffset(long offset); + + public abstract Builder setFinal(boolean isFinal); + + public abstract ChunkUploadRequest build(); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadResponse.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadResponse.java new file mode 100644 index 000000000000..26066a6b19de --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadResponse.java @@ -0,0 +1,80 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** + * Response value object representing the outcome of a chunk upload. + * + * @param response type of the upload operation + */ +@NullMarked +@BetaApi +@InternalApi +@AutoValue +public abstract class ChunkUploadResponse { + + /** Whether the overall resumable upload stream has finalized and completed on the server. */ + public abstract boolean isComplete(); + + /** + * The response object returned by the server upon final completion (e.g. metadata of the uploaded + * resource), or {@code null} if the upload is still in progress. + */ + public abstract @Nullable ResponseT getResponse(); + + public abstract Builder toBuilder(); + + public static Builder newBuilder() { + return new AutoValue_ChunkUploadResponse.Builder().setComplete(false); + } + + public static ChunkUploadResponse create( + boolean isComplete, @Nullable ResponseT response) { + return new AutoValue_ChunkUploadResponse.Builder() + .setComplete(isComplete) + .setResponse(response) + .build(); + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setComplete(boolean isComplete); + + public abstract Builder setResponse(@Nullable ResponseT response); + + public abstract ChunkUploadResponse build(); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/DataChunk.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/DataChunk.java new file mode 100644 index 000000000000..18658089a06a --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/DataChunk.java @@ -0,0 +1,60 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; + +/** + * A discrete chunk of data produced from an {@link java.io.InputStream} for a resumable upload, + * containing the binary payload, starting byte offset, and finality indicator. + */ +@NullMarked +@BetaApi +@InternalApi +@AutoValue +public abstract class DataChunk { + + /** The binary chunk payload. */ + @SuppressWarnings("mutable") + public abstract byte[] getPayload(); + + /** The byte offset where this chunk begins in the overall stream. */ + public abstract long getOffset(); + + /** Whether this chunk reaches the end of the input stream. */ + public abstract boolean isFinal(); + + public static DataChunk create(byte[] payload, long offset, boolean isFinal) { + return new AutoValue_DataChunk(payload, offset, isFinal); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusRequest.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusRequest.java new file mode 100644 index 000000000000..9c71b8ef3d38 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusRequest.java @@ -0,0 +1,50 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; + +/** Request value object for querying the status of an active resumable upload session. */ +@NullMarked +@BetaApi +@InternalApi +@AutoValue +public abstract class QueryStatusRequest { + + /** Returns the upload session URL to query. */ + public abstract String getUploadUrl(); + + public static QueryStatusRequest create(String uploadUrl) { + return new AutoValue_QueryStatusRequest(uploadUrl); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java new file mode 100644 index 000000000000..1b44f353af3a --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java @@ -0,0 +1,84 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** + * Response value object representing the status and committed offset of a resumable upload session. + * + * @param response type of the upload operation + */ +@NullMarked +@BetaApi +@InternalApi +@AutoValue +public abstract class QueryStatusResponse { + + /** + * The total number of bytes successfully received and committed by the server so far, or {@code + * null} if the server did not return a committed offset (e.g. if the upload is already + * finalized). + * + *

When {@link #isComplete()} is {@code false}, this value is guaranteed to be non-null and + * represents the starting offset for resuming the upload. + */ + public abstract @Nullable Long getCommittedOffset(); + + /** Whether the resumable upload session has finalized and completed on the server. */ + public abstract boolean isComplete(); + + /** + * The response object returned by the server upon final completion (e.g. metadata of the uploaded + * resource), or {@code null} if the upload is still in progress. + */ + public abstract @Nullable ResponseT getResponse(); + + public abstract Builder toBuilder(); + + public static Builder newBuilder() { + return new AutoValue_QueryStatusResponse.Builder().setComplete(false); + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setCommittedOffset(@Nullable Long committedOffset); + + public abstract Builder setComplete(boolean isComplete); + + public abstract Builder setResponse(@Nullable ResponseT response); + + public abstract QueryStatusResponse build(); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadClient.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadClient.java index d8913a1ab659..daeb94df6a67 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadClient.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadClient.java @@ -29,6 +29,7 @@ */ package com.google.api.gax.resumable; +import com.google.api.core.BetaApi; import com.google.api.core.InternalApi; import com.google.api.gax.rpc.UnaryCallable; import org.jspecify.annotations.NullMarked; @@ -40,9 +41,16 @@ * @param response type of the upload operation */ @NullMarked +@BetaApi @InternalApi public interface ResumableUploadClient { /** Returns a {@link UnaryCallable} to initiate a resumable upload session. */ UnaryCallable startUploadCallable(); + + /** Returns a {@link UnaryCallable} to transmit an individual chunk. */ + UnaryCallable> uploadChunkCallable(); + + /** Returns a {@link UnaryCallable} to query the status and offset of an active upload session. */ + UnaryCallable> queryStatusCallable(); } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java new file mode 100644 index 000000000000..6389a43df42a --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java @@ -0,0 +1,47 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import org.jspecify.annotations.NullMarked; + +/** A callback listener for observing the progress and state transitions of a resumable upload. */ +@BetaApi +@FunctionalInterface +@NullMarked +public interface ResumableUploadProgressListener { + + /** + * Invoked when upload progress or state changes. + * + * @param status the current status snapshot of the upload + */ + void onProgress(ResumableUploadStatus status); +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java index d6d82d8e7b0f..345432957aa8 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java @@ -29,12 +29,14 @@ */ package com.google.api.gax.resumable; +import com.google.api.core.BetaApi; import com.google.api.core.InternalApi; import com.google.auto.value.AutoValue; import org.jspecify.annotations.NullMarked; /** Represents the session metadata returned after starting a resumable upload. */ @NullMarked +@BetaApi @InternalApi @AutoValue public abstract class ResumableUploadSession { diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadStatus.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadStatus.java new file mode 100644 index 000000000000..b822200d7c56 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadStatus.java @@ -0,0 +1,98 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** Status snapshot of an ongoing or completed resumable upload session. */ +@BetaApi +@NullMarked +@AutoValue +public abstract class ResumableUploadStatus { + + /** The state of the resumable upload session. */ + public enum State { + /** Session initiation is in progress (acquiring upload session URL). */ + STARTING, + + /** Transmitting chunk payloads to the server. */ + UPLOADING, + + /** A recoverable error occurred; querying server status and resynchronizing offset. */ + RECOVERING, + + /** The server query status succeeded and the committed offset was received. */ + OFFSET_RECEIVED, + + /** The upload was successfully finalized by the server. */ + FINALIZED, + + /** The upload failed unrecoverably or was cancelled. */ + FAILED + } + + /** + * Returns the negotiated upload session URI, or {@code null} if session initiation is pending. + */ + public abstract @Nullable String getUploadUrl(); + + /** Returns the number of bytes successfully uploaded to the server so far. */ + public abstract long getBytesUploaded(); + + /** Returns the current state of the upload session. */ + public abstract State getState(); + + /** Returns the exception that triggered recovery or caused failure, if any. */ + public abstract @Nullable Throwable getException(); + + public abstract Builder toBuilder(); + + public static Builder newBuilder() { + return new AutoValue_ResumableUploadStatus.Builder() + .setBytesUploaded(0L) + .setState(State.STARTING); + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setUploadUrl(@Nullable String uploadUrl); + + public abstract Builder setBytesUploaded(long bytesUploaded); + + public abstract Builder setState(State state); + + public abstract Builder setException(@Nullable Throwable exception); + + public abstract ResumableUploadStatus build(); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/RewindableStreamBuffer.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/RewindableStreamBuffer.java new file mode 100644 index 000000000000..0b70a1f0fac1 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/RewindableStreamBuffer.java @@ -0,0 +1,223 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; + +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.common.io.ByteStreams; +import com.google.errorprone.annotations.concurrent.GuardedBy; +import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** + * A stream buffer that maintains a sliding single-chunk window over an {@link InputStream} for + * resumable uploads, yielding self-describing {@link DataChunk} instances and supporting in-memory + * rewinds and forward seeking. + */ +@BetaApi +@InternalApi +@NullMarked +class RewindableStreamBuffer implements AutoCloseable { + + private final Object lock = new Object(); + private final InputStream source; + private final int chunkSize; + + // Retained in-memory byte chunk for transmission and intra-chunk rewind. + @GuardedBy("lock") + private byte @Nullable [] currentChunk; + + // Absolute byte offset in the stream where currentChunk begins. + @GuardedBy("lock") + private long chunkStartOffset = 0L; + + // Current read cursor indicating the start offset for nextChunk(). + @GuardedBy("lock") + private long cursorOffset = 0L; + + // Indicates whether the end of the underlying stream has been reached. + @GuardedBy("lock") + private boolean endOfStream = false; + + RewindableStreamBuffer(InputStream source, int chunkSize) { + this.source = checkNotNull(source, "source must not be null"); + checkArgument(chunkSize > 0, "chunkSize must be > 0"); + this.chunkSize = chunkSize; + } + + // Computes the physical byte position read from the underlying stream. + @GuardedBy("lock") + private long streamPosition() { + return chunkStartOffset + (currentChunk != null ? currentChunk.length : 0); + } + + /** + * Reads and returns the next {@link DataChunk} of up to {@code chunkSize} bytes. + * + * @return data chunk for transmission + * @throws IOException on error reading from the stream + */ + DataChunk nextChunk() throws IOException { + synchronized (lock) { + // 1. If cursorOffset is within the retained currentChunk (e.g. retry after partial failure) + if (currentChunk != null && cursorOffset < chunkStartOffset + currentChunk.length) { + int offsetInChunk = (int) (cursorOffset - chunkStartOffset); + int length = Math.min(chunkSize, currentChunk.length - offsetInChunk); + byte[] payload = + (offsetInChunk == 0 && length == currentChunk.length) + ? currentChunk + : Arrays.copyOfRange(currentChunk, offsetInChunk, offsetInChunk + length); + boolean isFinal = endOfStream && (offsetInChunk + length == currentChunk.length); + long chunkOffset = cursorOffset; + cursorOffset += length; + return DataChunk.create(payload, chunkOffset, isFinal); + } + + // 2. If already at EOF and no uncommitted bytes remain + if (endOfStream) { + return DataChunk.create(new byte[0], cursorOffset, true); + } + + // 3. Read up to chunkSize from the underlying stream + byte[] buffer = new byte[chunkSize]; + int totalRead = 0; + + while (totalRead < chunkSize) { + int read = source.read(buffer, totalRead, chunkSize - totalRead); + if (read == -1) { + endOfStream = true; + break; + } + totalRead += read; + } + + if (totalRead == 0 && endOfStream) { + return DataChunk.create(new byte[0], cursorOffset, true); + } + + byte[] payload = totalRead == chunkSize ? buffer : Arrays.copyOf(buffer, totalRead); + currentChunk = payload; + chunkStartOffset = cursorOffset; + cursorOffset += totalRead; + + boolean isFinal = endOfStream; + return DataChunk.create(payload, chunkStartOffset, isFinal); + } + } + + /** + * Confirms that the server has received bytes up to {@code serverCommittedOffset}. + * + *

Discards any buffered data prior to {@code serverCommittedOffset}. + * + * @param serverCommittedOffset newly confirmed committed offset + */ + void acknowledge(long serverCommittedOffset) { + synchronized (lock) { + if (currentChunk != null && serverCommittedOffset >= chunkStartOffset + currentChunk.length) { + currentChunk = null; + chunkStartOffset = serverCommittedOffset; + if (cursorOffset < serverCommittedOffset) { + cursorOffset = serverCommittedOffset; + } + } + } + } + + /** + * Re-aligns the buffer cursor to {@code serverCommittedOffset} following recovery or session + * resumption. + * + *

+ * + * @param serverCommittedOffset server's committed byte offset + * @throws IOException if the server offset rolled back before confirmed bytes + */ + void seekTo(long serverCommittedOffset) throws IOException { + checkArgument(serverCommittedOffset >= 0, "serverCommittedOffset must not be negative"); + synchronized (lock) { + // 1. Intra-chunk rewind: cursor moves within existing in-memory chunk + if (currentChunk != null + && serverCommittedOffset >= chunkStartOffset + && serverCommittedOffset <= chunkStartOffset + currentChunk.length) { + cursorOffset = serverCommittedOffset; + return; + } + + // 2. Forward seek (e.g. session resumption): discard buffer and skip underlying stream + long physicalPosition = streamPosition(); + if (serverCommittedOffset >= physicalPosition) { + currentChunk = null; + long bytesToSkip = serverCommittedOffset - physicalPosition; + if (bytesToSkip > 0) { + ByteStreams.skipFully(source, bytesToSkip); + } + chunkStartOffset = serverCommittedOffset; + cursorOffset = serverCommittedOffset; + return; + } + + // 3. Backward seek outside retained buffer: unrecoverable protocol error + throw new IOException( + String.format( + "Cannot rewind stream to offset %d (retained buffer start: %d). " + + "Server committed offset rolled back before confirmed bytes.", + serverCommittedOffset, chunkStartOffset)); + } + } + + /** Returns true if the end of the underlying stream has been reached. */ + boolean isEndOfStream() { + synchronized (lock) { + return endOfStream + && (currentChunk == null || cursorOffset >= chunkStartOffset + currentChunk.length); + } + } + + @Override + public void close() throws IOException { + synchronized (lock) { + currentChunk = null; + source.close(); + } + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFuture.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFuture.java index ea2605e3a5d7..4ba50fac8eeb 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFuture.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFuture.java @@ -31,6 +31,11 @@ import com.google.api.core.ApiFuture; import com.google.api.core.BetaApi; +import com.google.api.gax.resumable.ResumableUploadProgressListener; +import com.google.api.gax.resumable.ResumableUploadStatus; +import java.util.concurrent.Executor; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; /** * A specialized {@link ApiFuture} for tracking and controlling an in-flight resumable upload. @@ -38,8 +43,27 @@ * @param response type */ @BetaApi +@NullMarked public interface ResumableUploadFuture extends ApiFuture { /** Returns the upload session URL, or {@code null} if session initiation is in progress. */ - String getUploadSessionUrl(); + @Nullable String getUploadSessionUrl(); + + /** Returns the current status snapshot of the upload. */ + ResumableUploadStatus getStatus(); + + /** + * Registers a listener for progress updates on the direct executor. + * + * @param listener the listener to receive progress updates + */ + void addProgressListener(ResumableUploadProgressListener listener); + + /** + * Registers a listener for progress updates on the specified executor. + * + * @param listener the listener to receive progress updates + * @param executor the executor on which to run the listener + */ + void addProgressListener(ResumableUploadProgressListener listener, Executor executor); } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/ChunkUploadRequestTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/ChunkUploadRequestTest.java new file mode 100644 index 000000000000..a3f7f4d9a15b --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/ChunkUploadRequestTest.java @@ -0,0 +1,66 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import static com.google.common.truth.Truth.assertThat; + +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; + +class ChunkUploadRequestTest { + + @Test + void builder_defaultIsFinalFalse() { + byte[] payload = "test-payload".getBytes(StandardCharsets.UTF_8); + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl("https://upload.example.com/session/1") + .setPayload(payload) + .setOffset(0L) + .build(); + + assertThat(request.isFinal()).isFalse(); + assertThat(request.getPayload()).isEqualTo(payload); + } + + @Test + void builder_explicitIsFinalTrue_preservesValue() { + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl("https://upload.example.com/session/1") + .setPayload(new byte[0]) + .setOffset(1024L) + .setFinal(true) + .build(); + + assertThat(request.isFinal()).isTrue(); + assertThat(request.getPayload()).isEmpty(); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/ResumableUploadStatusTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/ResumableUploadStatusTest.java new file mode 100644 index 000000000000..071471ed84e9 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/ResumableUploadStatusTest.java @@ -0,0 +1,73 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import static com.google.common.truth.Truth.assertThat; + +import org.junit.jupiter.api.Test; + +class ResumableUploadStatusTest { + + @Test + void builder_defaults() { + ResumableUploadStatus status = ResumableUploadStatus.newBuilder().build(); + assertThat(status.getUploadUrl()).isNull(); + assertThat(status.getBytesUploaded()).isEqualTo(0L); + assertThat(status.getState()).isEqualTo(ResumableUploadStatus.State.STARTING); + assertThat(status.getException()).isNull(); + } + + @Test + void builder_explicitValuesAndToBuilder() { + Exception ex = new RuntimeException("test error"); + ResumableUploadStatus status = + ResumableUploadStatus.newBuilder() + .setUploadUrl("https://upload.url/session-1") + .setBytesUploaded(1024L) + .setState(ResumableUploadStatus.State.UPLOADING) + .setException(ex) + .build(); + + assertThat(status.getUploadUrl()).isEqualTo("https://upload.url/session-1"); + assertThat(status.getBytesUploaded()).isEqualTo(1024L); + assertThat(status.getState()).isEqualTo(ResumableUploadStatus.State.UPLOADING); + assertThat(status.getException()).isSameInstanceAs(ex); + + ResumableUploadStatus modified = + status.toBuilder() + .setState(ResumableUploadStatus.State.FINALIZED) + .setBytesUploaded(2048L) + .build(); + + assertThat(modified.getState()).isEqualTo(ResumableUploadStatus.State.FINALIZED); + assertThat(modified.getBytesUploaded()).isEqualTo(2048L); + assertThat(modified.getUploadUrl()).isEqualTo("https://upload.url/session-1"); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/RewindableStreamBufferTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/RewindableStreamBufferTest.java new file mode 100644 index 000000000000..ab269911648e --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/resumable/RewindableStreamBufferTest.java @@ -0,0 +1,218 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; + +class RewindableStreamBufferTest { + + @Test + void testReadSequentialChunks() throws IOException { + byte[] data = "HelloWorld123456".getBytes(StandardCharsets.UTF_8); // 16 bytes + ByteArrayInputStream stream = new ByteArrayInputStream(data); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(stream, 8); + + // Read first chunk (8 bytes: "HelloWor") + DataChunk chunk1 = buffer.nextChunk(); + assertThat(new String(chunk1.getPayload(), StandardCharsets.UTF_8)).isEqualTo("HelloWor"); + assertThat(chunk1.getOffset()).isEqualTo(0); + assertThat(chunk1.isFinal()).isFalse(); + + // Acknowledge first chunk + buffer.acknowledge(8); + + // Read second chunk (8 bytes: "ld123456") + DataChunk chunk2 = buffer.nextChunk(); + assertThat(new String(chunk2.getPayload(), StandardCharsets.UTF_8)).isEqualTo("ld123456"); + assertThat(chunk2.getOffset()).isEqualTo(8); + assertThat(chunk2.isFinal()).isFalse(); + + // Subsequent read returns empty final chunk + buffer.acknowledge(16); + DataChunk chunk3 = buffer.nextChunk(); + assertThat(chunk3.getPayload()).isEmpty(); + assertThat(chunk3.getOffset()).isEqualTo(16); + assertThat(chunk3.isFinal()).isTrue(); + + buffer.close(); + } + + @Test + void testSeekAndRewindWithinChunk() throws IOException { + byte[] data = "0123456789ABCDEF".getBytes(StandardCharsets.UTF_8); // 16 bytes + ByteArrayInputStream stream = new ByteArrayInputStream(data); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(stream, 8); + + // Read first chunk of 8 bytes ("01234567") + DataChunk chunk1 = buffer.nextChunk(); + assertThat(new String(chunk1.getPayload(), StandardCharsets.UTF_8)).isEqualTo("01234567"); + assertThat(chunk1.getOffset()).isEqualTo(0); + assertThat(chunk1.isFinal()).isFalse(); + + // Simulate failure where server only committed 3 bytes ("012") + buffer.seekTo(3); + + // Retry reading from offset 3 (should return slice "34567" from retained buffer) + DataChunk retrySlice = buffer.nextChunk(); + assertThat(new String(retrySlice.getPayload(), StandardCharsets.UTF_8)).isEqualTo("34567"); + assertThat(retrySlice.getOffset()).isEqualTo(3); + assertThat(retrySlice.isFinal()).isFalse(); + + // Once server commits full 8 bytes + buffer.acknowledge(8); + + // Read next chunk ("89ABCDEF") + DataChunk chunk2 = buffer.nextChunk(); + assertThat(new String(chunk2.getPayload(), StandardCharsets.UTF_8)).isEqualTo("89ABCDEF"); + assertThat(chunk2.getOffset()).isEqualTo(8); + assertThat(chunk2.isFinal()).isFalse(); + + // Subsequent read returns empty final chunk + buffer.acknowledge(16); + DataChunk chunk3 = buffer.nextChunk(); + assertThat(chunk3.getPayload()).isEmpty(); + assertThat(chunk3.getOffset()).isEqualTo(16); + assertThat(chunk3.isFinal()).isTrue(); + + buffer.close(); + } + + @Test + void testPartialReadSmallerThanChunkSize() throws IOException { + byte[] data = "Small".getBytes(StandardCharsets.UTF_8); // 5 bytes + ByteArrayInputStream stream = new ByteArrayInputStream(data); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(stream, 10); + + DataChunk chunk = buffer.nextChunk(); + assertThat(new String(chunk.getPayload(), StandardCharsets.UTF_8)).isEqualTo("Small"); + assertThat(chunk.getOffset()).isEqualTo(0); + assertThat(chunk.isFinal()).isTrue(); + + buffer.acknowledge(5); + DataChunk nextChunk = buffer.nextChunk(); + assertThat(nextChunk.getPayload()).isEmpty(); + assertThat(nextChunk.getOffset()).isEqualTo(5); + assertThat(nextChunk.isFinal()).isTrue(); + + buffer.close(); + } + + @Test + void testEmptyStream() throws IOException { + ByteArrayInputStream stream = new ByteArrayInputStream(new byte[0]); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(stream, 8); + + DataChunk chunk = buffer.nextChunk(); + assertThat(chunk.getPayload()).isEmpty(); + assertThat(chunk.getOffset()).isEqualTo(0); + assertThat(chunk.isFinal()).isTrue(); + + buffer.close(); + } + + @Test + void testSeekFromBeginning() throws IOException { + byte[] data = "0123456789ABCDEF".getBytes(StandardCharsets.UTF_8); // 16 bytes + ByteArrayInputStream stream = new ByteArrayInputStream(data); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(stream, 8); + + // Seek directly to offset 8 before any reads (e.g. resuming session) + buffer.seekTo(8); + + DataChunk chunk = buffer.nextChunk(); + assertThat(new String(chunk.getPayload(), StandardCharsets.UTF_8)).isEqualTo("89ABCDEF"); + assertThat(chunk.getOffset()).isEqualTo(8); + assertThat(chunk.isFinal()).isFalse(); + + buffer.acknowledge(16); + DataChunk finalChunk = buffer.nextChunk(); + assertThat(finalChunk.getPayload()).isEmpty(); + assertThat(finalChunk.getOffset()).isEqualTo(16); + assertThat(finalChunk.isFinal()).isTrue(); + + buffer.close(); + } + + @Test + void testSeekForwardAcrossChunks() throws IOException { + byte[] data = "0123456789ABCDEF".getBytes(StandardCharsets.UTF_8); // 16 bytes + ByteArrayInputStream stream = new ByteArrayInputStream(data); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(stream, 4); + + DataChunk chunk1 = buffer.nextChunk(); + assertThat(new String(chunk1.getPayload(), StandardCharsets.UTF_8)).isEqualTo("0123"); + assertThat(chunk1.getOffset()).isEqualTo(0); + assertThat(chunk1.isFinal()).isFalse(); + + // Seek past chunk 1 to offset 10 + buffer.seekTo(10); + + DataChunk chunk2 = buffer.nextChunk(); + assertThat(new String(chunk2.getPayload(), StandardCharsets.UTF_8)).isEqualTo("ABCD"); + assertThat(chunk2.getOffset()).isEqualTo(10); + assertThat(chunk2.isFinal()).isFalse(); + + buffer.close(); + } + + @Test + void testConstructorValidatesChunkSize() { + ByteArrayInputStream stream = new ByteArrayInputStream(new byte[0]); + assertThrows(IllegalArgumentException.class, () -> new RewindableStreamBuffer(stream, 0)); + assertThrows(IllegalArgumentException.class, () -> new RewindableStreamBuffer(stream, -1)); + } + + @Test + void testCannotSeekBackwardsBeforeChunkStartThrowsIOException() throws IOException { + byte[] data = "0123456789ABCDEF".getBytes(StandardCharsets.UTF_8); + ByteArrayInputStream stream = new ByteArrayInputStream(data); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(stream, 8); + + // Read first chunk (0..8) and acknowledge it + DataChunk chunk1 = buffer.nextChunk(); + assertThat(chunk1.getOffset()).isEqualTo(0); + buffer.acknowledge(8); + + // Read second chunk (8..16) + DataChunk chunk2 = buffer.nextChunk(); + assertThat(chunk2.getOffset()).isEqualTo(8); + + // Attempting to seek back before chunkStartOffset (8) throws IOException + assertThrows(IOException.class, () -> buffer.seekTo(4)); + + buffer.close(); + } +}