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 index ee3c9f82e9e4..227a576749bb 100644 --- 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 @@ -38,6 +38,8 @@ import com.google.api.core.SettableApiFuture; import com.google.api.gax.resumable.ChunkUploadRequest; import com.google.api.gax.resumable.ChunkUploadResponse; +import com.google.api.gax.resumable.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.resumable.StartUploadRequest; @@ -86,6 +88,9 @@ public final class HttpJsonResumableUploadClient implements ResumableUploadClien private static final PathTemplate PATH_TEMPLATE = PathTemplate.create("{+path}"); + private static final Map> QUERY_STATUS_HEADERS = + ImmutableMap.of(UPLOAD_COMMAND_HEADER, ImmutableList.of("query")); + private static final ApiMethodDescriptor START_UPLOAD_DESCRIPTOR = ApiMethodDescriptor.newBuilder() .setFullMethodName("ResumableUpload/StartUpload") @@ -155,6 +160,41 @@ public PathTemplate getPathTemplate() { .setResponseParser(StringHttpResponseParser.create()) .build(); + 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 HttpContent getHttpContent(QueryStatusRequest request) { + return new EmptyContent(); + } + + @Override + public String getPath(QueryStatusRequest request) { + return request.getUploadUrl(); + } + + @Override + public PathTemplate getPathTemplate() { + return PATH_TEMPLATE; + } + }) + .setResponseParser(StringHttpResponseParser.create()) + .build(); + private final ClientContext clientContext; public static HttpJsonResumableUploadClient create(ClientContext clientContext) { @@ -230,6 +270,32 @@ public ApiFuture futureCall( }; } + @Override + public UnaryCallable queryStatusCallable() { + return new UnaryCallable() { + @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); + + SettableApiFuture future = SettableApiFuture.create(); + HttpJsonClientCalls.startUnaryCall( + clientCall, request, context, new QueryStatusResponseListener(future)); + + return future; + } + }; + } + private static class StartUploadResponseListener extends HttpJsonClientCall.Listener { private final SettableApiFuture future; @@ -308,22 +374,10 @@ private static class ChunkUploadResponseListener extends HttpJsonClientCall.List @Override public void onHeaders(HttpJsonMetadata responseHeaders) { - Map headers = responseHeaders.getHeaders(); - - String statusStr = HttpHeadersUtils.getFirstHeader(headers, UPLOAD_STATUS_HEADER); - if (STATUS_FINAL.equalsIgnoreCase(statusStr)) { - this.isComplete = true; - } - - String sizeReceivedStr = - HttpHeadersUtils.getFirstHeader(headers, UPLOAD_SIZE_RECEIVED_HEADER); - if (!Strings.isNullOrEmpty(sizeReceivedStr)) { - try { - this.committedOffset = Long.parseLong(sizeReceivedStr); - } catch (NumberFormatException ignored) { - // Ignore invalid/malformed size received header and fall back to local offset - // calculation. - } + this.isComplete = isUploadFinal(responseHeaders); + Long sizeReceived = parseSizeReceived(responseHeaders); + if (sizeReceived != null) { + this.committedOffset = sizeReceived; } } @@ -359,6 +413,83 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) { } } + private static class QueryStatusResponseListener extends HttpJsonClientCall.Listener { + + private final SettableApiFuture future; + private boolean isComplete = false; + @Nullable private Long committedOffset = null; + private String responseBody = ""; + + QueryStatusResponseListener(SettableApiFuture future) { + this.future = future; + } + + @Override + public void onHeaders(HttpJsonMetadata responseHeaders) { + this.isComplete = isUploadFinal(responseHeaders); + this.committedOffset = parseSizeReceived(responseHeaders); + } + + @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 (isComplete || committedOffset != null) { + future.set( + QueryStatusResponse.create( + committedOffset != null ? committedOffset : 0L, + isComplete, + isComplete ? responseBody : "")); + } else { + future.setException( + ApiExceptionFactory.createException( + "Query status response did not contain valid X-Goog-Upload-Size-Received header", + /* cause= */ null, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false)); + } + } else { + future.setException( + createApiException(statusCode, trailers, "Failed to query upload status")); + } + } catch (Throwable t) { + future.setException( + ApiExceptionFactory.createException( + "Internal error processing query status response", + t, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false)); + } + } + } + + private static boolean isUploadFinal(HttpJsonMetadata responseHeaders) { + String statusStr = + HttpHeadersUtils.getFirstHeader(responseHeaders.getHeaders(), UPLOAD_STATUS_HEADER); + return STATUS_FINAL.equalsIgnoreCase(statusStr); + } + + @Nullable + private static Long parseSizeReceived(HttpJsonMetadata responseHeaders) { + String sizeReceivedStr = + HttpHeadersUtils.getFirstHeader(responseHeaders.getHeaders(), UPLOAD_SIZE_RECEIVED_HEADER); + if (!Strings.isNullOrEmpty(sizeReceivedStr)) { + try { + return Long.parseLong(sizeReceivedStr); + } catch (NumberFormatException ignored) { + // Unparseable header; return null and let the listener decide how to handle it. + } + } + return null; + } + private static ApiException createApiException( int statusCode, @Nullable HttpJsonMetadata trailers, String actionDescription) { Throwable cause = trailers != null ? trailers.getException() : null; 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 index dbdfcee45ab3..d663c65ff34c 100644 --- 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 @@ -40,6 +40,8 @@ import com.google.api.client.testing.http.MockLowLevelHttpResponse; 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.resumable.StartUploadRequest; import com.google.api.gax.rpc.AbortedException; @@ -504,4 +506,220 @@ void uploadChunk_serverReturnsErrorWithoutException_throwsApiException() { assertThat(exception.getCause()).isInstanceOf(ApiException.class); assertThat(exception.getCause()).hasMessageThat().contains("500"); } + + @Test + void queryStatus_activeUpload_returnsCommittedOffset() { + Map> capturedHeaders = new HashMap<>(); + String[] capturedUrl = new String[1]; + + HttpTransport httpTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + capturedUrl[0] = url; + return new MockLowLevelHttpRequest() { + @Override + public LowLevelHttpResponse execute() { + capturedHeaders.putAll(getHeaders()); + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + response.addHeader("X-Goog-Upload-Status", "active"); + response.addHeader("X-Goog-Upload-Size-Received", "524288"); + return response; + } + }; + } + }; + + HttpJsonResumableUploadClient client = createClient(httpTransport); + QueryStatusRequest request = + QueryStatusRequest.create("https://test.googleapis.com/upload/session/123"); + + QueryStatusResponse response = client.queryStatusCallable().call(request); + + assertThat(response.isComplete()).isFalse(); + assertThat(response.getCommittedOffset()).isEqualTo(524288L); + assertThat(response.getResponseBody()).isEmpty(); + + assertThat(capturedUrl[0]).contains("https://test.googleapis.com/upload/session/123"); + assertThat(capturedHeaders).containsKey("x-goog-upload-command"); + assertThat(capturedHeaders.get("x-goog-upload-command")).contains("query"); + } + + @Test + void queryStatus_finalUpload_returnsCompleteAndResponseBody() { + Map> capturedHeaders = new HashMap<>(); + String[] capturedUrl = new String[1]; + + HttpTransport httpTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + capturedUrl[0] = url; + return new MockLowLevelHttpRequest() { + @Override + public LowLevelHttpResponse execute() { + capturedHeaders.putAll(getHeaders()); + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + response.addHeader("X-Goog-Upload-Status", "final"); + response.addHeader("X-Goog-Upload-Size-Received", "1048576"); + response.setContent("{\"name\":\"uploaded-file.txt\",\"size\":1048576}"); + return response; + } + }; + } + }; + + HttpJsonResumableUploadClient client = createClient(httpTransport); + QueryStatusRequest request = + QueryStatusRequest.create("https://test.googleapis.com/upload/session/123"); + + QueryStatusResponse response = client.queryStatusCallable().call(request); + + assertThat(response.isComplete()).isTrue(); + assertThat(response.getCommittedOffset()).isEqualTo(1048576L); + assertThat(response.getResponseBody()) + .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}"); + + assertThat(capturedHeaders).containsKey("x-goog-upload-command"); + assertThat(capturedHeaders.get("x-goog-upload-command")).contains("query"); + } + + @Test + void queryStatus_finalUploadWithoutSizeReceivedHeader_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("https://test.googleapis.com/upload/session/123"); + + QueryStatusResponse response = client.queryStatusCallable().call(request); + + assertThat(response.isComplete()).isTrue(); + assertThat(response.getCommittedOffset()).isEqualTo(0L); + assertThat(response.getResponseBody()) + .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}"); + } + + @Test + void queryStatus_withCustomExtraHeaders_preservesHeaders() { + Map> capturedHeaders = new HashMap<>(); + + HttpTransport httpTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest() { + @Override + public LowLevelHttpResponse execute() { + capturedHeaders.putAll(getHeaders()); + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + response.addHeader("X-Goog-Upload-Status", "active"); + response.addHeader("X-Goog-Upload-Size-Received", "256"); + return response; + } + }; + } + }; + + HttpJsonResumableUploadClient client = createClient(httpTransport); + QueryStatusRequest request = + QueryStatusRequest.create("https://test.googleapis.com/upload/session/123"); + + Map> customHeaders = new HashMap<>(); + customHeaders.put("X-Custom-Query-Header", Collections.singletonList("CustomQueryValue")); + + ApiCallContext callContext = + HttpJsonCallContext.createDefault().withExtraHeaders(customHeaders); + + QueryStatusResponse response = client.queryStatusCallable().call(request, callContext); + + assertThat(response.getCommittedOffset()).isEqualTo(256L); + assertThat(capturedHeaders).containsKey("x-custom-query-header"); + assertThat(capturedHeaders.get("x-custom-query-header")).contains("CustomQueryValue"); + assertThat(capturedHeaders).containsKey("x-goog-upload-command"); + assertThat(capturedHeaders.get("x-goog-upload-command")).contains("query"); + } + + @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_missingSizeReceivedHeader_throwsException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "active"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + QueryStatusRequest request = + QueryStatusRequest.create("https://test.googleapis.com/upload/session/123"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.queryStatusCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(InternalException.class); + assertThat(exception.getCause()) + .hasMessageThat() + .contains("Query status response did not contain valid X-Goog-Upload-Size-Received header"); + } + + @Test + void queryStatus_malformedSizeReceivedHeader_throwsException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "active"); + httpResponse.addHeader("X-Goog-Upload-Size-Received", "not-a-number"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + QueryStatusRequest request = + QueryStatusRequest.create("https://test.googleapis.com/upload/session/123"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.queryStatusCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(InternalException.class); + assertThat(exception.getCause()) + .hasMessageThat() + .contains("Query status response did not contain valid X-Goog-Upload-Size-Received header"); + } + + @Test + void queryStatus_serverReturnsErrorWithoutException_throwsApiException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(500); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + QueryStatusRequest request = + QueryStatusRequest.create("https://test.googleapis.com/upload/session/123"); + + ExecutionException exception = + assertThrows( + ExecutionException.class, () -> client.queryStatusCallable().futureCall(request).get()); + + assertThat(exception.getCause()).isInstanceOf(ApiException.class); + assertThat(exception.getCause()).hasMessageThat().contains("500"); + } } 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..9acda075793a --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusRequest.java @@ -0,0 +1,48 @@ +/* + * 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.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 +@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..71005dcaa887 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java @@ -0,0 +1,67 @@ +/* + * 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.InternalApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; + +/** Response value object representing the status and committed offset of a resumable upload session. */ +@NullMarked +@InternalApi +@AutoValue +public abstract class QueryStatusResponse { + + /** + * The total number of bytes successfully received and committed by the server so far. + * + *

This value is the starting offset for resuming the upload. + */ + public abstract long getCommittedOffset(); + + /** Whether the resumable upload session has finalized and completed on the server. */ + public abstract boolean isComplete(); + + /** + * The response body returned by the server upon final completion (e.g. JSON metadata of the + * uploaded resource), or an empty string if no body was returned or the upload is still in + * progress. + */ + public abstract String getResponseBody(); + + public static QueryStatusResponse create(long committedOffset, boolean isComplete) { + return create(committedOffset, isComplete, ""); + } + + public static QueryStatusResponse create( + long committedOffset, boolean isComplete, String responseBody) { + return new AutoValue_QueryStatusResponse(committedOffset, isComplete, responseBody); + } +} 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 7087239ee544..28e0a29c7edc 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 @@ -43,4 +43,7 @@ public interface ResumableUploadClient { /** 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(); }