diff --git a/sdk-platform-java/gax-java/RESUMABLE_UPLOAD_DESIGN.md b/sdk-platform-java/gax-java/RESUMABLE_UPLOAD_DESIGN.md new file mode 100644 index 000000000000..03301858b0f2 --- /dev/null +++ b/sdk-platform-java/gax-java/RESUMABLE_UPLOAD_DESIGN.md @@ -0,0 +1,263 @@ +# Resumable Upload Protocol for Java: Design Document + +This document proposes the architecture and design for integrating the Resumable Upload Protocol (RUP) into `gax-java` and the GAPIC code generator. + +--- + +## 1. Design Principles and Requirements + +1. **Veneer and GAPIC Aligned**: The solution must integrate cleanly into the existing `Callable` framework of `gax-java`. +2. **Stream-Safe Retries**: Java `InputStream` is forward-only. The design must provide a clean abstraction (`InputStreamProvider`) to recreate or seek the stream during recovery. +3. **Double-Loop Retry & Recovery**: Implements the precise Category 1 (transient) and Category 2 (state consistency) error classification with backoffs as described in the RUP specifications. +4. **Progress Reporting**: Supports asynchronous progress updates via a simple callback mechanism. +5. **No unnecessary chunking**: By default, uploads send the remaining bytes in one request (avoiding unnecessary memory buffering or chunk management). + +--- + +## 2. API Design (`gax` Changes) + +We introduce a new callable type and request/response wrappers in `com.google.api.gax.rpc`. + +### 2.1. `InputStreamProvider` + +To support seeking/rewinding, the stream source is wrapped in a functional interface that can supply fresh streams on retry: + +```java +package com.google.api.gax.rpc; + +import java.io.IOException; +import java.io.InputStream; + +/** Provides a fresh {@link InputStream} for retriable upload operations. */ +@FunctionalInterface +public interface InputStreamProvider { + /** Returns a new {@link InputStream}. */ + InputStream get() throws IOException; +} +``` + +### 2.2. Progress Listener and Status + +```java +package com.google.api.gax.rpc; + +/** Listener for tracking progress of a resumable upload. */ +public interface ResumableUploadProgressListener { + + enum State { + NOT_STARTED, + IN_PROGRESS, + RECOVERING, + COMPLETED, + FAILED, + CANCELLED + } + + void onProgress(ResumableUploadStatus status); +} + +/** Status details for progress updates. */ +public final class ResumableUploadStatus { + private final long bytesUploaded; + private final long totalBytes; + private final ResumableUploadProgressListener.State state; + + public ResumableUploadStatus(long bytesUploaded, long totalBytes, ResumableUploadProgressListener.State state) { + this.bytesUploaded = bytesUploaded; + this.totalBytes = totalBytes; + this.state = state; + } + + public long getBytesUploaded() { return bytesUploaded; } + public long getTotalBytes() { return totalBytes; } + public ResumableUploadProgressListener.State getState() { return state; } +} +``` + +### 2.3. Request Wrapper: `ResumableUploadRequest` + +```java +package com.google.api.gax.rpc; + +import com.google.common.base.Preconditions; + +public final class ResumableUploadRequest { + private final RequestT request; + private final InputStreamProvider streamProvider; + private final long totalBytes; // -1 if unknown + private final ResumableUploadProgressListener progressListener; + + private ResumableUploadRequest(Builder builder) { + this.request = Preconditions.checkNotNull(builder.request); + this.streamProvider = Preconditions.checkNotNull(builder.streamProvider); + this.totalBytes = builder.totalBytes; + this.progressListener = builder.progressListener; + } + + public RequestT getRequest() { return request; } + public InputStreamProvider getStreamProvider() { return streamProvider; } + public long getTotalBytes() { return totalBytes; } + public ResumableUploadProgressListener getProgressListener() { return progressListener; } + + public static Builder newBuilder() { + return new Builder<>(); + } + + public static class Builder { + private RequestT request; + private InputStreamProvider streamProvider; + private long totalBytes = -1; + private ResumableUploadProgressListener progressListener; + + public Builder setRequest(RequestT request) { + this.request = request; + return this; + } + public Builder setStreamProvider(InputStreamProvider streamProvider) { + this.streamProvider = streamProvider; + return this; + } + public Builder setTotalBytes(long totalBytes) { + this.totalBytes = totalBytes; + return this; + } + public Builder setProgressListener(ResumableUploadProgressListener progressListener) { + this.progressListener = progressListener; + return this; + } + public ResumableUploadRequest build() { + return new ResumableUploadRequest<>(this); + } + } +} +``` + +### 2.4. Callable Wrapper: `ResumableUploadCallable` + +```java +package com.google.api.gax.rpc; + +import com.google.api.core.ApiFuture; + +public abstract class ResumableUploadCallable { + + protected ResumableUploadCallable() {} + + public abstract ApiFuture futureCall( + ResumableUploadRequest request, ApiCallContext context); + + public ResponseT call(ResumableUploadRequest request, ApiCallContext context) { + return ApiExceptions.callAndTranslateCharSequenceException(futureCall(request, context)); + } + + public ResponseT call(ResumableUploadRequest request) { + return call(request, null); + } +} +``` + +--- + +## 3. Transport Implementation (`gax-httpjson`) + +The transport layer executes the actual HTTP protocol calls using the Google HTTP Client. + +We introduce `HttpJsonResumableUploadCall` to coordinate the resumable upload state machine. + +### 3.1. Error Categorization in Java + +```java +private enum ErrorCategory { + CATEGORY_1_TRANSIENT, // 429, 500, 502, 503, 504, TCP/Socket Timeout + CATEGORY_2_MISMATCH, // 400, 412, 416 + CATEGORY_3_FATAL // 401, 403, 404, etc. +} + +private ErrorCategory getErrorCategory(Throwable t) { + if (t instanceof HttpResponseException) { + int statusCode = ((HttpResponseException) t).getStatusCode(); + if (statusCode == 429 || statusCode >= 500) { + return ErrorCategory.CATEGORY_1_TRANSIENT; + } + if (statusCode == 400 || statusCode == 412 || statusCode == 416) { + return ErrorCategory.CATEGORY_2_MISMATCH; + } + } + if (t instanceof IOException) { + // Socket timeouts, connection drops + return ErrorCategory.CATEGORY_1_TRANSIENT; + } + return ErrorCategory.CATEGORY_3_FATAL; +} +``` + +### 3.2. Detailed Execution Flow (State Machine) + +The `HttpJsonResumableUploadCall` runs inside the user's thread (or client executor pool for future execution) and implements the following flow: + +```mermaid +stateDiagram-v2 + [*] --> StartSession + StartSession --> UploadLoop : Success (200 OK + active) + StartSession --> StartSession : Cat 1 Transient (Backoff) + StartSession --> [*] : Cat 3 Fatal / Deadline Exceeded + + state UploadLoop { + [*] --> OpenStream + OpenStream --> SkipToOffset + SkipToOffset --> TransmitChunk + TransmitChunk --> [*] : Success (final) + TransmitChunk --> QueryState : Cat 2 Mismatch / Socket Drop + TransmitChunk --> TransmitChunk : Cat 1 Transient (Backoff) + } + + UploadLoop --> [*] : Success + UploadLoop --> [*] : Cat 3 Fatal / Deadline Exceeded + + state QueryState { + [*] --> SendQuery + SendQuery --> ResumeUpload : Success (active + new offset) + SendQuery --> [*] : Success (final) + SendQuery --> SendQuery : Cat 1 Transient (Backoff) + SendQuery --> [*] : Cat 3 Fatal + } + + QueryState --> UploadLoop : Resume +``` + +#### Step 1: Start Session +- Build standard headers + merge user-provided metadata. +- Pre-emptively prefix headers that affect physical bodies (`Content-Length`, `Content-Type`, etc.) with `X-Goog-Upload-Header-`. +- Set `X-Goog-Upload-Protocol: resumable` and `X-Goog-Upload-Command: start`. +- Execute POST with the request JSON body. +- Extract `X-Goog-Upload-URL` header value to obtain the `uploadUrl`. +- Extract `X-Goog-Upload-Chunk-Granularity` and adjust the user-configured `chunkSize` to the largest multiple of this granularity value (rounded down, minimum equal to granularity). + +#### Step 2: Upload Loop (Transmit) +- Check absolute global deadline. +- Retrieve the chunk corresponding to the current `offset`: + - Search in the 2-chunk memory cache. + - If not found: + - Check if the underlying stream position matches `offset`. + - If not, close and recreate the stream from `streamProvider.get()` and skip/seek to `offset`. + - Read `adjustedChunkSize` bytes from the stream, store as a `BufferedChunk`, add to the cache (retaining at most 2 chunks), and update the stream position. +- If no data was read (stream reached EOF at a chunk boundary): + - Send an empty POST request with `X-Goog-Upload-Command: finalize`. +- If data was read: + - If it is the last chunk (length < chunk size): + - Send a POST request with `X-Goog-Upload-Command: upload, finalize` and `X-Goog-Upload-Offset: offset`. + - If it is an intermediate chunk: + - Send a POST request with `X-Goog-Upload-Command: upload` and `X-Goog-Upload-Offset: offset`. +- Update the progress listener upon successful responses. +- If the server replies with status `final` (even on `upload` command): parse the response and complete the upload. +- If exception occurs: Categorize exception. If Category 2 (Mismatch) or socket drop, transition to **Query State**. + +#### Step 3: Query State +- Execute POST to `uploadUrl` with `X-Goog-Upload-Command: query`. +- If response is `final`: parse the response and complete the upload (via `UploadAlreadyFinalizedException` handling). +- If response is `active`: + - Extract `X-Goog-Upload-Size-Received` -> `newOffset`. + - If `newOffset == offset`: apply backoff (to avoid spamming server). + - Update `offset = newOffset` and transition back to **Upload Loop** (which will automatically retrieve the correct chunk from cache or recreate and seek the stream). +- If Category 1 (Transient) error: retry query with backoff. +- If Category 3 (Fatal) error: fail immediately. diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallableFactory.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallableFactory.java index 447fc46dd9e0..898e83f2c86a 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallableFactory.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallableFactory.java @@ -39,6 +39,7 @@ import com.google.api.gax.rpc.OperationCallSettings; import com.google.api.gax.rpc.OperationCallable; import com.google.api.gax.rpc.PagedCallSettings; +import com.google.api.gax.rpc.ResumableUploadCallable; import com.google.api.gax.rpc.ServerStreamingCallSettings; import com.google.api.gax.rpc.ServerStreamingCallable; import com.google.api.gax.rpc.UnaryCallSettings; @@ -220,6 +221,20 @@ ServerStreamingCallable createServerStreamingCallable( return callable.withDefaultCallContext(clientContext.getDefaultCallContext()); } + /** + * Create a resumable upload callable object. Designed for use by generated code. + * + * @param httpJsonCallSettings the http/json call settings + * @param clientContext {@link ClientContext} to use to connect to the service. + * @return {@link ResumableUploadCallable} callable object. + */ + public static + ResumableUploadCallable createResumableUploadCallable( + HttpJsonCallSettings httpJsonCallSettings, + ClientContext clientContext) { + return new HttpJsonResumableUploadCallable<>(httpJsonCallSettings, clientContext); + } + static ApiTracerContext getApiTracerContext(@Nonnull ApiMethodDescriptor methodDescriptor) { return ApiTracerContext.newBuilder() .setFullMethodName(methodDescriptor.getFullMethodName()) diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCall.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCall.java new file mode 100644 index 000000000000..69aa096318ec --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCall.java @@ -0,0 +1,730 @@ +/* + * 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.ByteArrayContent; +import com.google.api.client.http.EmptyContent; +import com.google.api.client.http.GenericUrl; +import com.google.api.client.http.HttpContent; +import com.google.api.client.http.HttpMediaType; +import com.google.api.client.http.HttpRequest; +import com.google.api.client.http.HttpRequestFactory; +import com.google.api.client.http.HttpResponse; +import com.google.api.client.http.HttpResponseException; +import com.google.api.client.http.HttpTransport; +import com.google.api.client.http.json.JsonHttpContent; +import com.google.api.client.json.JsonFactory; +import com.google.api.client.json.gson.GsonFactory; +import com.google.api.client.util.GenericData; +import com.google.api.core.ApiFuture; +import com.google.api.core.SettableApiFuture; +import com.google.api.gax.rpc.DeadlineExceededException; +import com.google.api.gax.rpc.ResumableUploadProgressListener; +import com.google.api.gax.rpc.ResumableUploadRequest; +import com.google.api.gax.rpc.ResumableUploadStatus; +import com.google.auth.Credentials; +import com.google.auth.http.HttpCredentialsAdapter; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executor; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** Encapsulates the execution logic and state machine of the Resumable Upload protocol. */ +final class HttpJsonResumableUploadCall { + + private static final Logger logger = + Logger.getLogger(HttpJsonResumableUploadCall.class.getName()); + private static final JsonFactory JSON_FACTORY = GsonFactory.getDefaultInstance(); + private static final int HTTP_BAD_REQUEST = 400; + private static final int HTTP_TOO_MANY_REQUESTS = 429; + private static final int HTTP_INTERNAL_ERROR = 500; + private static final int HTTP_PRECONDITION_FAILED = 412; + private static final int HTTP_RANGE_NOT_SATISFIABLE = 416; + private static final long DEFAULT_BACKOFF_BASE_MS = 500; + private static final long DEFAULT_BACKOFF_MAX_MS = 30000; + private static final int DEFAULT_DEADLINE_MINUTES = 10; + + private final ApiMethodDescriptor methodDescriptor; + private final ResumableUploadRequest uploadRequest; + private final HttpTransport httpTransport; + private final HttpJsonMetadata requestHeaders; + private final HttpJsonCallOptions callOptions; + private final String endpoint; + private final Executor executor; + + private enum ErrorCategory { + CATEGORY_1_TRANSIENT, + CATEGORY_2_MISMATCH, + CATEGORY_3_FATAL + } + + HttpJsonResumableUploadCall( + ApiMethodDescriptor methodDescriptor, + ResumableUploadRequest uploadRequest, + HttpTransport httpTransport, + HttpJsonMetadata requestHeaders, + HttpJsonCallOptions callOptions, + String endpoint, + Executor executor) { + this.methodDescriptor = Preconditions.checkNotNull(methodDescriptor); + this.uploadRequest = Preconditions.checkNotNull(uploadRequest); + this.httpTransport = Preconditions.checkNotNull(httpTransport); + this.requestHeaders = Preconditions.checkNotNull(requestHeaders); + this.callOptions = Preconditions.checkNotNull(callOptions); + this.endpoint = Preconditions.checkNotNull(endpoint); + this.executor = Preconditions.checkNotNull(executor); + } + + ApiFuture execute() { + SettableApiFuture future = SettableApiFuture.create(); + executor.execute( + () -> { + try { + ResponseT result = runStateMachine(); + future.set(result); + } catch (Throwable t) { + future.setException(t); + } + }); + return future; + } + + private ResponseT runStateMachine() throws Exception { + try { + return runStateMachineInternal(); + } catch (HttpResponseException e) { + throw translateException(e); + } + } + + private static class BufferedChunk { + final long offset; + final byte[] data; + final int length; + + BufferedChunk(long offset, byte[] data, int length) { + this.offset = offset; + this.data = data; + this.length = length; + } + } + + private static class SessionInfo { + final String uploadUrl; + final int granularity; + + SessionInfo(String uploadUrl, int granularity) { + this.uploadUrl = uploadUrl; + this.granularity = granularity; + } + } + + private static final class UploadAlreadyFinalizedException extends Exception { + private final Object response; + + UploadAlreadyFinalizedException(Object response) { + this.response = response; + } + + Object getResponse() { + return response; + } + } + + private ResponseT runStateMachineInternal() throws Exception { + Instant deadline = calculateDeadline(); + + // Phase 1: Start Session (with retry) + String uploadUrl = null; + int attempt = 0; + int granularity = 1; + while (true) { + try { + checkDeadline(deadline); + SessionInfo sessionInfo = startSession(deadline); + uploadUrl = sessionInfo.uploadUrl; + granularity = sessionInfo.granularity; + break; // Success + } catch (Exception e) { + checkDeadline(deadline); + ErrorCategory category = getErrorCategory(e); + if (category == ErrorCategory.CATEGORY_1_TRANSIENT) { + attempt++; + long delayMs = calculateBackoff(attempt); + logger.log( + Level.WARNING, + "Transient error starting session. Backing off for " + delayMs + " ms", + e); + sleep(delayMs); + } else { + throw e; // Fatal/Mismatch, bubble up + } + } + } + logger.log( + Level.FINE, + "Resumable session started. Upload URL: {0}, Granularity: {1}", + new Object[] {uploadUrl, granularity}); + + int adjustedChunkSize = uploadRequest.getChunkSize(); + if (granularity > 1) { + adjustedChunkSize = (adjustedChunkSize / granularity) * granularity; + if (adjustedChunkSize == 0) { + adjustedChunkSize = granularity; + } + } + + long offset = 0; + attempt = 0; + long previousOffset = -1; + + InputStream stream = uploadRequest.getStreamProvider().get(); + long streamPosition = 0; + + List cache = new ArrayList<>(); + + // Phase 2 & 3 Loop: Transmit Chunks & Query Recovery + while (true) { + try { + checkDeadline(deadline); + + // Find chunk in cache or read from stream + BufferedChunk chunk = null; + for (BufferedChunk cached : cache) { + if (cached.offset == offset) { + chunk = cached; + break; + } + } + + if (chunk == null) { + // Read from stream + if (streamPosition != offset) { + if (stream != null) { + stream.close(); + } + stream = uploadRequest.getStreamProvider().get(); + long skipped = skipFully(stream, offset); + if (skipped < offset) { + throw new IOException("Failed to skip stream bytes to offset: " + offset); + } + streamPosition = offset; + } + + byte[] buffer = new byte[adjustedChunkSize]; + int bytesRead = readFully(stream, buffer, adjustedChunkSize); + if (bytesRead > 0) { + chunk = new BufferedChunk(offset, buffer, bytesRead); + cache.add(chunk); + if (cache.size() > 2) { + cache.remove(0); + } + streamPosition += bytesRead; + } + } + + if (chunk == null) { + // Stream was empty or exact chunk multiple and fully uploaded. + // Send finalize only + return sendFinalizeOnly(uploadUrl, offset, deadline); + } + + // Check if this is the last chunk + boolean isEof = (chunk.length < adjustedChunkSize); + + if (isEof) { + // Send upload, finalize for the last chunk + return sendUploadFinalize(uploadUrl, chunk.offset, chunk.data, chunk.length, deadline); + } + + // Send intermediate chunk (upload command) + sendChunk(uploadUrl, chunk.offset, chunk.data, chunk.length, deadline); + + // Successful chunk transmission! Update offset to next chunk + offset = chunk.offset + chunk.length; + attempt = 0; // Reset backoff attempts on progress + + } catch (UploadAlreadyFinalizedException uafe) { + updateProgress( + uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset, + ResumableUploadProgressListener.State.COMPLETED); + return (ResponseT) uafe.getResponse(); + } catch (Exception e) { + checkDeadline(deadline); + ErrorCategory category = getErrorCategory(e); + + if (category == ErrorCategory.CATEGORY_2_MISMATCH) { + logger.log(Level.WARNING, "State mismatch detected. Triggering recovery...", e); + updateProgress(offset, ResumableUploadProgressListener.State.RECOVERING); + + long serverOffset = 0; + try { + serverOffset = recoverOffset(uploadUrl, deadline); + } catch (UploadAlreadyFinalizedException uafe) { + updateProgress( + uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset, + ResumableUploadProgressListener.State.COMPLETED); + return (ResponseT) uafe.getResponse(); + } + + logger.log(Level.INFO, "Recovery completed. Server received bytes: {0}", serverOffset); + + if (serverOffset == previousOffset) { + attempt++; + long delayMs = calculateBackoff(attempt); + sleep(delayMs); + } else { + attempt = 0; + previousOffset = serverOffset; + } + + offset = serverOffset; + // Loop will handle finding the chunk in cache or seeking/recreating the stream! + } else if (category == ErrorCategory.CATEGORY_1_TRANSIENT) { + attempt++; + long delayMs = calculateBackoff(attempt); + logger.log( + Level.WARNING, + "Transient error. Backing off for {0} ms (attempt {1})", + new Object[] {delayMs, attempt}); + sleep(delayMs); + } else { + updateProgress(offset, ResumableUploadProgressListener.State.FAILED); + throw e; // Fatal, bubble up + } + } + } + } + + private SessionInfo startSession(final Instant deadline) throws Exception { + HttpRequestFactory requestFactory = getRequestFactory(); + HttpRequestFormatter requestFormatter = methodDescriptor.getRequestFormatter(); + + GenericData tokenRequest = new GenericData(); + String requestBody = requestFormatter.getRequestBody(uploadRequest.getRequest()); + HttpContent initialContent; + + if (!Strings.isNullOrEmpty(requestBody)) { + JSON_FACTORY.createJsonParser(requestBody).parse(tokenRequest); + initialContent = + new JsonHttpContent(JSON_FACTORY, tokenRequest) + .setMediaType(new HttpMediaType("application/json; charset=utf-8")); + } else { + initialContent = new EmptyContent(); + } + + String path = "/resumable/upload" + requestFormatter.getPath(uploadRequest.getRequest()); + GenericUrl url = new GenericUrl(normalizeEndpoint(endpoint) + path); + + Map> queryParams = + requestFormatter.getQueryParamNames(uploadRequest.getRequest()); + for (Map.Entry> queryParam : queryParams.entrySet()) { + if (queryParam.getValue() != null) { + url.set(queryParam.getKey(), queryParam.getValue()); + } + } + + HttpRequest httpRequest = requestFactory.buildPostRequest(url, initialContent); + configureTimeouts(httpRequest, deadline); + + for (Map.Entry entry : requestHeaders.getHeaders().entrySet()) { + String key = entry.getKey(); + String value = (String) entry.getValue(); + + if (isMetadataHeaderDenylisted(key)) { + httpRequest.getHeaders().set("X-Goog-Upload-Header-" + key, value); + } else { + httpRequest.getHeaders().set(key, value); + } + } + + httpRequest.getHeaders().set("X-Goog-Upload-Protocol", "resumable"); + httpRequest.getHeaders().set("X-Goog-Upload-Command", "start"); + + updateProgress(0, ResumableUploadProgressListener.State.NOT_STARTED); + + HttpResponse response = null; + try { + response = httpRequest.execute(); + String status = response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Status"); + if (!"active".equalsIgnoreCase(status)) { + throw new HttpResponseException.Builder( + response.getStatusCode(), response.getStatusMessage(), response.getHeaders()) + .setMessage("Failed to initiate resumable session: Status is not active") + .build(); + } + String uploadUrl = response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-URL"); + if (Strings.isNullOrEmpty(uploadUrl)) { + throw new HttpResponseException.Builder( + response.getStatusCode(), response.getStatusMessage(), response.getHeaders()) + .setMessage("Failed to initiate resumable session: Missing upload URL") + .build(); + } + + String granularityStr = + response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Chunk-Granularity"); + int granularity = 1; + if (!Strings.isNullOrEmpty(granularityStr)) { + try { + granularity = Integer.parseInt(granularityStr); + } catch (NumberFormatException e) { + logger.log( + Level.WARNING, "Failed to parse chunk granularity header: " + granularityStr, e); + } + } + return new SessionInfo(uploadUrl, granularity); + } finally { + if (response != null) { + response.disconnect(); + } + } + } + + private void sendChunk( + final String uploadUrl, + final long offset, + final byte[] data, + final int length, + final Instant deadline) + throws Exception { + HttpRequestFactory requestFactory = getRequestFactory(); + HttpContent payload = new ByteArrayContent("application/octet-stream", data, 0, length); + + GenericUrl url = new GenericUrl(uploadUrl); + HttpRequest httpRequest = requestFactory.buildPostRequest(url, payload); + configureTimeouts(httpRequest, deadline); + + httpRequest.getHeaders().set("X-Goog-Upload-Command", "upload"); + httpRequest.getHeaders().set("X-Goog-Upload-Offset", String.valueOf(offset)); + + updateProgress(offset, ResumableUploadProgressListener.State.IN_PROGRESS); + + HttpResponse response = null; + try { + response = httpRequest.execute(); + String status = response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Status"); + if ("final".equalsIgnoreCase(status)) { + InputStreamReader reader = + new InputStreamReader(response.getContent(), StandardCharsets.UTF_8); + ResponseT parsedResponse = + methodDescriptor.getResponseParser().parse(reader, callOptions.getTypeRegistry()); + throw new UploadAlreadyFinalizedException(parsedResponse); + } + if (!"active".equalsIgnoreCase(status)) { + throw new HttpResponseException.Builder( + response.getStatusCode(), response.getStatusMessage(), response.getHeaders()) + .setMessage("Resumable upload chunk failed: Status is not active") + .build(); + } + } finally { + if (response != null) { + response.disconnect(); + } + } + } + + private ResponseT sendUploadFinalize( + final String uploadUrl, + final long offset, + final byte[] data, + final int length, + final Instant deadline) + throws Exception { + HttpRequestFactory requestFactory = getRequestFactory(); + HttpContent payload = new ByteArrayContent("application/octet-stream", data, 0, length); + + GenericUrl url = new GenericUrl(uploadUrl); + HttpRequest httpRequest = requestFactory.buildPostRequest(url, payload); + configureTimeouts(httpRequest, deadline); + + httpRequest.getHeaders().set("X-Goog-Upload-Command", "upload, finalize"); + httpRequest.getHeaders().set("X-Goog-Upload-Offset", String.valueOf(offset)); + + updateProgress(offset, ResumableUploadProgressListener.State.IN_PROGRESS); + + HttpResponse response = null; + try { + response = httpRequest.execute(); + String status = response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Status"); + if (!"final".equalsIgnoreCase(status)) { + throw new HttpResponseException.Builder( + response.getStatusCode(), response.getStatusMessage(), response.getHeaders()) + .setMessage("Resumable upload finalize failed: Status is not final") + .build(); + } + + InputStreamReader reader = + new InputStreamReader(response.getContent(), StandardCharsets.UTF_8); + ResponseT parsedResponse = + methodDescriptor.getResponseParser().parse(reader, callOptions.getTypeRegistry()); + + long finalProgressBytes = + uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset + length; + updateProgress(finalProgressBytes, ResumableUploadProgressListener.State.COMPLETED); + + return parsedResponse; + } finally { + if (response != null) { + response.disconnect(); + } + } + } + + private ResponseT sendFinalizeOnly( + final String uploadUrl, final long offset, final Instant deadline) throws Exception { + HttpRequestFactory requestFactory = getRequestFactory(); + + GenericUrl url = new GenericUrl(uploadUrl); + HttpRequest httpRequest = requestFactory.buildPostRequest(url, new EmptyContent()); + configureTimeouts(httpRequest, deadline); + + httpRequest.getHeaders().set("X-Goog-Upload-Command", "finalize"); + + updateProgress(offset, ResumableUploadProgressListener.State.IN_PROGRESS); + + HttpResponse response = null; + try { + response = httpRequest.execute(); + String status = response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Status"); + if (!"final".equalsIgnoreCase(status)) { + throw new HttpResponseException.Builder( + response.getStatusCode(), response.getStatusMessage(), response.getHeaders()) + .setMessage("Resumable upload finalize failed: Status is not final") + .build(); + } + + InputStreamReader reader = + new InputStreamReader(response.getContent(), StandardCharsets.UTF_8); + ResponseT parsedResponse = + methodDescriptor.getResponseParser().parse(reader, callOptions.getTypeRegistry()); + + long finalProgressBytes = + uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset; + updateProgress(finalProgressBytes, ResumableUploadProgressListener.State.COMPLETED); + + return parsedResponse; + } finally { + if (response != null) { + response.disconnect(); + } + } + } + + private long recoverOffset(final String uploadUrl, final Instant deadline) throws Exception { + HttpRequestFactory requestFactory = getRequestFactory(); + GenericUrl url = new GenericUrl(uploadUrl); + + HttpRequest httpRequest = requestFactory.buildPostRequest(url, new EmptyContent()); + configureTimeouts(httpRequest, deadline); + + httpRequest.getHeaders().set("X-Goog-Upload-Command", "query"); + + HttpResponse response = null; + try { + response = httpRequest.execute(); + String status = response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Status"); + + if ("final".equalsIgnoreCase(status)) { + InputStreamReader reader = + new InputStreamReader(response.getContent(), StandardCharsets.UTF_8); + ResponseT parsedResponse = + methodDescriptor.getResponseParser().parse(reader, callOptions.getTypeRegistry()); + throw new UploadAlreadyFinalizedException(parsedResponse); + } + if (!"active".equalsIgnoreCase(status)) { + throw new HttpResponseException.Builder( + response.getStatusCode(), response.getStatusMessage(), response.getHeaders()) + .setMessage("Query failed: Status is not active") + .build(); + } + + String receivedSizeStr = + response.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Size-Received"); + if (Strings.isNullOrEmpty(receivedSizeStr)) { + throw new HttpResponseException.Builder( + response.getStatusCode(), response.getStatusMessage(), response.getHeaders()) + .setMessage("Query failed: Missing size received header") + .build(); + } + + return Long.parseLong(receivedSizeStr); + } finally { + if (response != null) { + response.disconnect(); + } + } + } + + private int readFully(final InputStream in, final byte[] b, final int len) throws IOException { + int total = 0; + while (total < len) { + int result = in.read(b, total, len - total); + if (result == -1) { + break; + } + total += result; + } + return total == 0 && len > 0 ? -1 : total; + } + + private long skipFully(final InputStream in, final long n) throws IOException { + long total = 0; + while (total < n) { + long skipped = in.skip(n - total); + if (skipped == 0) { + int read = in.read(); + if (read == -1) { + break; + } + skipped = 1; + } + total += skipped; + } + return total; + } + + private void configureTimeouts(final HttpRequest request, final Instant deadline) { + long remainingMs = Duration.between(Instant.now(), deadline).toMillis(); + if (remainingMs <= 0) { + remainingMs = 1; // force timeout + } + request.setConnectTimeout((int) remainingMs); + request.setReadTimeout((int) remainingMs); + } + + private Instant calculateDeadline() { + Duration timeout = callOptions.getTimeoutDuration(); + if (timeout != null && !timeout.isZero() && !timeout.isNegative()) { + return Instant.now().plus(timeout); + } + return Instant.now().plus(Duration.ofMinutes(DEFAULT_DEADLINE_MINUTES)); + } + + private void checkDeadline(final Instant deadline) throws DeadlineExceededException { + if (Instant.now().isAfter(deadline)) { + throw (DeadlineExceededException) + com.google.api.gax.rpc.ApiExceptionFactory.createException( + "Resumable upload session exceeded the configured deadline.", + null, + HttpJsonStatusCode.of(com.google.api.gax.rpc.StatusCode.Code.DEADLINE_EXCEEDED), + false); + } + } + + private ErrorCategory getErrorCategory(final Throwable t) { + if (t instanceof HttpResponseException) { + HttpResponseException e = (HttpResponseException) t; + int statusCode = e.getStatusCode(); + String uploadStatus = e.getHeaders().getFirstHeaderStringValue("X-Goog-Upload-Status"); + if ("final".equalsIgnoreCase(uploadStatus)) { + return ErrorCategory.CATEGORY_3_FATAL; + } + if (statusCode == HTTP_TOO_MANY_REQUESTS || statusCode >= HTTP_INTERNAL_ERROR) { + return ErrorCategory.CATEGORY_1_TRANSIENT; + } + if (statusCode == HTTP_BAD_REQUEST + || statusCode == HTTP_PRECONDITION_FAILED + || statusCode == HTTP_RANGE_NOT_SATISFIABLE) { + return ErrorCategory.CATEGORY_2_MISMATCH; + } + return ErrorCategory.CATEGORY_3_FATAL; + } + if (t instanceof IOException) { + return ErrorCategory.CATEGORY_1_TRANSIENT; + } + return ErrorCategory.CATEGORY_3_FATAL; + } + + private long calculateBackoff(final int attempt) { + long baseDelay = DEFAULT_BACKOFF_BASE_MS; + long maxDelay = DEFAULT_BACKOFF_MAX_MS; + long delay = (long) (baseDelay * Math.pow(2, attempt)); + return Math.min(delay, maxDelay); + } + + private void sleep(final long ms) { + try { + Thread.sleep(ms); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private void updateProgress( + final long bytesUploaded, final ResumableUploadProgressListener.State state) { + ResumableUploadProgressListener progressListener = uploadRequest.getProgressListener(); + if (progressListener != null) { + progressListener.onProgress( + new ResumableUploadStatus(bytesUploaded, uploadRequest.getTotalBytes(), state)); + } + } + + private HttpRequestFactory getRequestFactory() { + Credentials credentials = callOptions.getCredentials(); + if (credentials != null) { + return httpTransport.createRequestFactory(new HttpCredentialsAdapter(credentials)); + } + return httpTransport.createRequestFactory(); + } + + private boolean isMetadataHeaderDenylisted(final String key) { + return "Content-Length".equalsIgnoreCase(key) + || "Content-Type".equalsIgnoreCase(key) + || "Content-Encoding".equalsIgnoreCase(key) + || "Transfer-Encoding".equalsIgnoreCase(key); + } + + private String normalizeEndpoint(final String rawEndpoint) { + String normalized = rawEndpoint; + if (!normalized.contains("://")) { + normalized = "https://" + normalized; + } + if (normalized.charAt(normalized.length() - 1) != '/') { + normalized += '/'; + } + return normalized; + } + + private Exception translateException(final HttpResponseException e) { + HttpJsonApiExceptionFactory factory = + new HttpJsonApiExceptionFactory(java.util.Collections.emptySet()); + return factory.create(e); + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCallable.java new file mode 100644 index 000000000000..84ef8c7a2b65 --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCallable.java @@ -0,0 +1,103 @@ +/* + * 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.HttpTransport; +import com.google.api.core.ApiFuture; +import com.google.api.core.BetaApi; +import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.ResumableUploadCallable; +import com.google.api.gax.rpc.ResumableUploadRequest; +import com.google.common.base.Preconditions; +import java.util.concurrent.Executor; + +/** + * A {@link ResumableUploadCallable} that uses HTTP/JSON transport. + * + * @param request type + * @param response type + */ +@BetaApi +public final class HttpJsonResumableUploadCallable + extends ResumableUploadCallable { + + private final HttpJsonCallSettings httpJsonCallSettings; + private final ClientContext clientContext; + + public HttpJsonResumableUploadCallable( + HttpJsonCallSettings httpJsonCallSettings, ClientContext clientContext) { + this.httpJsonCallSettings = Preconditions.checkNotNull(httpJsonCallSettings); + this.clientContext = Preconditions.checkNotNull(clientContext); + } + + @Override + public ApiFuture futureCall( + ResumableUploadRequest request, ApiCallContext context) { + Preconditions.checkNotNull(request); + + // Resolve call context + HttpJsonCallContext httpJsonContext = HttpJsonCallContext.createDefault(); + if (context != null) { + httpJsonContext = httpJsonContext.nullToSelf(context); + } + + // Resolve channel and endpoint + HttpJsonTransportChannel transportChannel = + (HttpJsonTransportChannel) clientContext.getTransportChannel(); + ManagedHttpJsonChannel channel = transportChannel.getManagedChannel(); + String endpoint = channel.getEndpoint(); + HttpTransport httpTransport = channel.getHttpTransport(); + + // Resolve credentials and executor + HttpJsonCallOptions callOptions = httpJsonContext.getCallOptions(); + if (callOptions.getCredentials() == null && clientContext.getCredentials() != null) { + callOptions = callOptions.toBuilder().setCredentials(clientContext.getCredentials()).build(); + } + + Executor executor = clientContext.getExecutor(); + + // Gather request headers + HttpJsonMetadata requestHeaders = + HttpJsonMetadata.newBuilder().build().withHeaders(httpJsonContext.getExtraHeaders()); + + HttpJsonResumableUploadCall call = + new HttpJsonResumableUploadCall<>( + httpJsonCallSettings.getMethodDescriptor(), + request, + httpTransport, + requestHeaders, + callOptions, + endpoint, + executor); + + return call.execute(); + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java index bd3bed855608..69dc242d6e8b 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java @@ -91,6 +91,10 @@ Executor getExecutor() { return executor; } + HttpTransport getHttpTransport() { + return httpTransport; + } + @Override public synchronized void shutdown() { // Calling shutdown/ shutdownNow() twice should no-op diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCallableTest.java b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCallableTest.java new file mode 100644 index 000000000000..cfe2863685ee --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadCallableTest.java @@ -0,0 +1,853 @@ +/* + * 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.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.gax.rpc.ApiException; +import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.InputStreamProvider; +import com.google.api.gax.rpc.ResumableUploadProgressListener; +import com.google.api.gax.rpc.ResumableUploadRequest; +import com.google.api.gax.rpc.ResumableUploadStatus; +import com.google.protobuf.TypeRegistry; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.Reader; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executors; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class HttpJsonResumableUploadCallableTest { + + @Mock private ApiMethodDescriptor methodDescriptor; + @Mock private HttpRequestFormatter requestFormatter; + @Mock private HttpResponseParser responseParser; + @Mock private ClientContext clientContext; + @Mock private HttpJsonTransportChannel transportChannel; + @Mock private ManagedHttpJsonChannel managedChannel; + + private java.util.concurrent.ScheduledExecutorService executor; + + @BeforeEach + void setUp() { + executor = Executors.newSingleThreadScheduledExecutor(); + Mockito.lenient().when(clientContext.getExecutor()).thenReturn(executor); + Mockito.lenient().when(clientContext.getTransportChannel()).thenReturn(transportChannel); + Mockito.lenient().when(transportChannel.getManagedChannel()).thenReturn(managedChannel); + Mockito.lenient().when(managedChannel.getEndpoint()).thenReturn("localhost"); + + // Wire formatter and parser mocking + Mockito.lenient().when(methodDescriptor.getRequestFormatter()).thenReturn(requestFormatter); + Mockito.lenient().when(methodDescriptor.getResponseParser()).thenReturn(responseParser); + Mockito.lenient() + .when(requestFormatter.getPath(Mockito.anyString())) + .thenReturn("/upload/resource"); + Mockito.lenient() + .when(requestFormatter.getRequestBody(Mockito.anyString())) + .thenReturn("{\"metadata\":\"value\"}"); + Mockito.lenient() + .when(requestFormatter.getQueryParamNames(Mockito.anyString())) + .thenReturn(Collections.emptyMap()); + } + + @AfterEach + void tearDown() { + executor.shutdown(); + } + + @Test + void happyPathUpload() throws Exception { + byte[] data = "Hello, World! Resumable upload".getBytes(); + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + // Sequence of mock HTTP responses + Queue mockResponses = new ConcurrentLinkedQueue<>(); + + // 1. Session start response + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + + // 2. Finalize upload response + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"success\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + MockLowLevelHttpResponse response = mockResponses.poll(); + if (response == null) { + throw new IOException("Unexpected out-of-bounds mock request: " + url); + } + return response; + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("SUCCESS_RESPONSE"); + + List progressHistory = new ArrayList<>(); + ResumableUploadProgressListener listener = progressHistory::add; + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setTotalBytes(data.length) + .setProgressListener(listener) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + + assertThat(response).isEqualTo("SUCCESS_RESPONSE"); + assertThat(progressHistory).isNotEmpty(); + + // Verify progress tracking states + assertThat(progressHistory.get(0).getState()) + .isEqualTo(ResumableUploadProgressListener.State.NOT_STARTED); + + ResumableUploadStatus lastStatus = progressHistory.get(progressHistory.size() - 1); + assertThat(lastStatus.getState()).isEqualTo(ResumableUploadProgressListener.State.COMPLETED); + assertThat(lastStatus.getBytesUploaded()).isEqualTo(data.length); + } + + @Test + void retryOnTransientStartError() throws Exception { + byte[] data = "Short stream".getBytes(); + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + + // 1. Session start transient failure (503) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(503) + .setReasonPhrase("Service Unavailable") + .setContent("")); + + // 2. Retry start session success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + + // 3. Finalize upload success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"ok\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("OK_REP"); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + assertThat(response).isEqualTo("OK_REP"); + } + + @Test + void stateMismatchRecoveryAndResume() throws Exception { + byte[] data = "First segment of data... Second segment of data".getBytes(); + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + + // 1. Session start success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + + // 2. Transmit failure with Category 2 (400 Bad Request) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(400) + .setReasonPhrase("Bad Request - offset mismatch") + .setContent("")); + + // 3. Query offset command (server has received 24 bytes) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-Size-Received", "24")); + + // 4. Upload rest starting from offset 24 success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"restored\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("RECOVERY_REP"); + + List progressHistory = new ArrayList<>(); + ResumableUploadProgressListener listener = progressHistory::add; + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setTotalBytes(data.length) + .setProgressListener(listener) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + + assertThat(response).isEqualTo("RECOVERY_REP"); + + // Verify recovery state was logged + boolean hasRecoveringState = false; + for (ResumableUploadStatus status : progressHistory) { + if (status.getState() == ResumableUploadProgressListener.State.RECOVERING) { + hasRecoveringState = true; + break; + } + } + assertThat(hasRecoveringState).isTrue(); + } + + @Test + void fatalErrorFailsImmediately() { + byte[] data = "test data".getBytes(); + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + + // 1. Session start returns 403 Forbidden (Category 3 Fatal) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(403) + .setReasonPhrase("Forbidden") + .setContent("")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + assertThrows(ApiException.class, () -> callable.call(request)); + } + + @Test + void chunkedUploadHappyPath() throws Exception { + byte[] data = "0123456789012345678901234".getBytes(); // 25 bytes + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + // 1. Session start + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + // 2. Chunk 1 (0-10) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 3. Chunk 2 (10-20) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 4. Chunk 3 (20-25) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"chunked_success\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("SUCCESS_CHUNKED"); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setChunkSize(10) + .setTotalBytes(data.length) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + assertThat(response).isEqualTo("SUCCESS_CHUNKED"); + } + + @Test + void chunkedUploadWithTransientError() throws Exception { + byte[] data = "012345678901234".getBytes(); // 15 bytes + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + // 1. Session start + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + // 2. Chunk 1 (0-10) -> Transient 503 error + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(503) + .setReasonPhrase("Service Unavailable") + .setContent("")); + // 3. Retry Chunk 1 (0-10) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 4. Chunk 2 (10-15) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"transient_retry_success\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("SUCCESS_TRANSIENT"); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setChunkSize(10) + .setTotalBytes(data.length) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + assertThat(response).isEqualTo("SUCCESS_TRANSIENT"); + } + + @Test + void chunkedUploadWithMismatchRecoveryInMemory() throws Exception { + byte[] data = "0123456789012345678901234".getBytes(); // 25 bytes + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + // 1. Session start + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + // 2. Chunk 1 (0-10) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 3. Chunk 2 (10-20) -> Mismatch 400 error + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(400) + .setReasonPhrase("Bad Request") + .setContent("")); + // 4. Query offset -> Server returns 10 (mismatch recovery offset matches currentChunkOffset) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-Size-Received", "10")); + // 5. Retry Chunk 2 (10-20) from memory buffer -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 6. Chunk 3 (20-25) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"mismatch_success\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("SUCCESS_MISMATCH"); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setChunkSize(10) + .setTotalBytes(data.length) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + assertThat(response).isEqualTo("SUCCESS_MISMATCH"); + } + + @Test + void chunkedUploadWithRecoveryFromPreviousChunk() throws Exception { + byte[] data = "0123456789012345678901234".getBytes(); // 25 bytes + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + // 1. Session start + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + // 2. Chunk 1 (0-10) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 3. Chunk 2 (10-20) -> Mismatch 400 error + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(400) + .setReasonPhrase("Bad Request") + .setContent("")); + // 4. Query offset -> Server returns 0 (which matches previousChunkOffset) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-Size-Received", "0")); + // 5. Resend Chunk 1 (0-10) from memory buffer -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 6. Send Chunk 2 (10-20) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 7. Chunk 3 (20-25) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"prev_chunk_success\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("SUCCESS_PREV_CHUNK"); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setChunkSize(10) + .setTotalBytes(data.length) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + assertThat(response).isEqualTo("SUCCESS_PREV_CHUNK"); + } + + @Test + void chunkedUploadWithRecoveryBySeekingStream() throws Exception { + byte[] data = "01234567890123456789012345678901234".getBytes(); // 35 bytes + java.util.concurrent.atomic.AtomicInteger streamCreationCount = + new java.util.concurrent.atomic.AtomicInteger(0); + InputStreamProvider streamProvider = + () -> { + streamCreationCount.incrementAndGet(); + return new ByteArrayInputStream(data); + }; + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + // 1. Session start + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345")); + // 2. Chunk 1 (0-10) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 3. Chunk 2 (10-20) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 4. Chunk 3 (20-30) -> Mismatch 400 error + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(400) + .setReasonPhrase("Bad Request") + .setContent("")); + // 5. Query offset -> Server returns 5 (outside the 2-chunk buffer: previous offset was 10, + // current is 20) + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-Size-Received", "5")); + // 6. Client seeks stream to 5. Sends upload (offset 5, length 10, i.e., bytes 5-15) -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 7. Sends chunk 15-25 -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 8. Sends chunk 25-35 -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"seek_success\"}")); + + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("SUCCESS_SEEK"); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setChunkSize(10) + .setTotalBytes(data.length) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + assertThat(response).isEqualTo("SUCCESS_SEEK"); + // Verify that the stream was recreated (once initially, and once on seek recovery) + assertThat(streamCreationCount.get()).isEqualTo(2); + } + + @Test + void granularityAlignment() throws Exception { + byte[] data = "01234567890123456789".getBytes(); // 20 bytes + InputStreamProvider streamProvider = () -> new ByteArrayInputStream(data); + + Queue mockResponses = new ConcurrentLinkedQueue<>(); + // 1. Session start with granularity constraint = 8 + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active") + .addHeader("X-Goog-Upload-URL", "https://localhost/session/12345") + .addHeader("X-Goog-Upload-Chunk-Granularity", "8")); + // 2. First chunk: adjusted chunk size = 8 (largest multiple of 8 <= 10). Length = 8. -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 3. Second chunk: length = 8 -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "active")); + // 4. Third chunk (last chunk): length = 4 -> Success + mockResponses.add( + new MockLowLevelHttpResponse() + .setStatusCode(200) + .addHeader("X-Goog-Upload-Status", "final") + .setContent("{\"response\":\"granularity_success\"}")); + + List requestCommandsAndOffsets = new ArrayList<>(); + MockHttpTransport transport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() throws IOException { + String cmd = getFirstHeaderValue("X-Goog-Upload-Command"); + String offset = getFirstHeaderValue("X-Goog-Upload-Offset"); + if (cmd != null) { + requestCommandsAndOffsets.add(cmd + ":" + offset); + } + return mockResponses.poll(); + } + }; + } + }; + + Mockito.when(managedChannel.getHttpTransport()).thenReturn(transport); + Mockito.when( + responseParser.parse(Mockito.any(Reader.class), Mockito.nullable(TypeRegistry.class))) + .thenReturn("SUCCESS_GRANULARITY"); + + ResumableUploadRequest request = + ResumableUploadRequest.newBuilder() + .setRequest("META") + .setStreamProvider(streamProvider) + .setChunkSize(10) // 10 is user-requested, will be aligned down to 8 + .setTotalBytes(data.length) + .build(); + + HttpJsonCallSettings settings = + HttpJsonCallSettings.newBuilder() + .setMethodDescriptor(methodDescriptor) + .build(); + + HttpJsonResumableUploadCallable callable = + new HttpJsonResumableUploadCallable<>(settings, clientContext); + + String response = callable.call(request); + assertThat(response).isEqualTo("SUCCESS_GRANULARITY"); + + // Verify command sequences and offsets: + // Chunk 1: upload:0 + // Chunk 2: upload:8 + // Chunk 3: upload, finalize:16 + assertThat(requestCommandsAndOffsets) + .containsExactly("start:null", "upload:0", "upload:8", "upload, finalize:16"); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/InputStreamProvider.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/InputStreamProvider.java new file mode 100644 index 000000000000..9d65114c5dac --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/InputStreamProvider.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.rpc; + +import com.google.api.core.BetaApi; +import java.io.IOException; +import java.io.InputStream; + +/** + * Provides a fresh {@link InputStream} for retriable upload operations. + * This is used to seek or rewind a stream when recovering from errors. + */ +@BetaApi +@FunctionalInterface +public interface InputStreamProvider { + /** + * Returns a new {@link InputStream}. + * + * @return a new input stream + * @throws IOException if the stream cannot be created + */ + InputStream get() throws IOException; +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java new file mode 100644 index 000000000000..5c3480fb7744 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java @@ -0,0 +1,87 @@ +/* + * 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.rpc; + +import com.google.api.core.ApiFuture; +import com.google.api.core.BetaApi; + +/** + * A ResumableUploadCallable is an API-transport-independent wrapper for the Resumable Upload + * protocol. + * + * @param request type + * @param response type + */ +@BetaApi +public abstract class ResumableUploadCallable { + + protected ResumableUploadCallable() {} + + /** + * Performs the resumable upload asynchronously. + * + * @param request the upload request options + * @param context the context of the call + * @return future for the response + */ + public abstract ApiFuture futureCall( + ResumableUploadRequest request, ApiCallContext context); + + /** + * Performs the resumable upload asynchronously. + * + * @param request the upload request options + * @return future for the response + */ + public ApiFuture futureCall(ResumableUploadRequest request) { + return futureCall(request, null); + } + + /** + * Performs the resumable upload synchronously. + * + * @param request the upload request options + * @param context the context of the call + * @return the RPC response + */ + public ResponseT call(ResumableUploadRequest request, ApiCallContext context) { + return ApiExceptions.callAndTranslateApiException(futureCall(request, context)); + } + + /** + * Performs the resumable upload synchronously. + * + * @param request the upload request options + * @return the RPC response + */ + public ResponseT call(ResumableUploadRequest request) { + return call(request, null); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressListener.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressListener.java new file mode 100644 index 000000000000..f843f209f396 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressListener.java @@ -0,0 +1,55 @@ +/* + * 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.rpc; + +import com.google.api.core.BetaApi; + +/** Listener for tracking the progress of a resumable upload session. */ +@BetaApi +@FunctionalInterface +public interface ResumableUploadProgressListener { + + /** The state of the upload session. */ + enum State { + NOT_STARTED, + IN_PROGRESS, + RECOVERING, + COMPLETED, + FAILED, + CANCELLED + } + + /** + * Invoked when upload progress or state changes. + * + * @param status the current status of the upload + */ + void onProgress(ResumableUploadStatus status); +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadRequest.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadRequest.java new file mode 100644 index 000000000000..b1435773bf64 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadRequest.java @@ -0,0 +1,210 @@ +/* + * 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.rpc; + +import com.google.api.core.BetaApi; +import com.google.common.base.Preconditions; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +/** + * Parameter class for a resumable upload call. + * Contains the request metadata, stream payload, and progress listener. + * + * @param the type of request message + */ +@BetaApi +public final class ResumableUploadRequest { + /** Default chunk size. */ + private static final int DEFAULT_CHUNK_SIZE = 10 * 1024 * 1024; + + /** Request metadata. */ + private final RequestT request; + /** Stream provider. */ + private final InputStreamProvider streamProvider; + /** Total bytes. */ + private final long totalBytes; + /** Chunk size. */ + private final int chunkSize; + /** Progress listener. */ + private final ResumableUploadProgressListener progressListener; + + private ResumableUploadRequest(final Builder builder) { + this.request = Preconditions.checkNotNull(builder.request); + this.streamProvider = Preconditions.checkNotNull(builder.streamProvider); + this.totalBytes = builder.totalBytes; + this.chunkSize = builder.chunkSize; + this.progressListener = builder.progressListener; + } + + /** + * Returns the metadata request message. + * + * @return the request metadata message + */ + @Nonnull + public RequestT getRequest() { + return request; + } + + /** + * Returns the stream provider. + * + * @return the stream provider + */ + @Nonnull + public InputStreamProvider getStreamProvider() { + return streamProvider; + } + + /** + * Returns the total size of the stream, or -1 if unknown. + * + * @return the total bytes + */ + public long getTotalBytes() { + return totalBytes; + } + + /** + * Returns the size of each upload chunk. + * + * @return the chunk size in bytes + */ + public int getChunkSize() { + return chunkSize; + } + + /** + * Returns the progress listener, or null if not set. + * + * @return the progress listener + */ + @Nullable + public ResumableUploadProgressListener getProgressListener() { + return progressListener; + } + + /** + * Creates a new builder for {@link ResumableUploadRequest}. + * + * @param type of the request + * @return a new builder + */ + public static Builder newBuilder() { + return new Builder<>(); + } + + /** + * Builder for {@link ResumableUploadRequest}. + * + * @param type of the request + */ + public static final class Builder { + /** Request metadata. */ + private RequestT request; + /** Stream provider. */ + private InputStreamProvider streamProvider; + /** Total bytes. */ + private long totalBytes = -1; + /** Chunk size. */ + private int chunkSize = DEFAULT_CHUNK_SIZE; + /** Progress listener. */ + private ResumableUploadProgressListener progressListener; + + /** + * Sets the request metadata. + * + * @param requestVal the request metadata + * @return the builder + */ + public Builder setRequest(final RequestT requestVal) { + this.request = requestVal; + return this; + } + + /** + * Sets the stream provider. + * + * @param streamProviderVal the stream provider + * @return the builder + */ + public Builder setStreamProvider( + final InputStreamProvider streamProviderVal) { + this.streamProvider = streamProviderVal; + return this; + } + + /** + * Sets the total size of the stream. + * + * @param totalBytesVal the total size, or -1 if unknown + * @return the builder + */ + public Builder setTotalBytes(final long totalBytesVal) { + this.totalBytes = totalBytesVal; + return this; + } + + /** + * Sets the size of each upload chunk. + * + * @param chunkSizeVal the chunk size in bytes + * @return the builder + */ + public Builder setChunkSize(final int chunkSizeVal) { + Preconditions.checkArgument( + chunkSizeVal > 0, "chunkSize must be greater than 0"); + this.chunkSize = chunkSizeVal; + return this; + } + + /** + * Sets the progress listener. + * + * @param progressListenerVal the progress listener + * @return the builder + */ + public Builder setProgressListener( + final ResumableUploadProgressListener progressListenerVal) { + this.progressListener = progressListenerVal; + return this; + } + + /** + * Builds a {@link ResumableUploadRequest}. + * + * @return the request + */ + public ResumableUploadRequest build() { + return new ResumableUploadRequest<>(this); + } + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.java new file mode 100644 index 000000000000..e5c73670d422 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.java @@ -0,0 +1,74 @@ +/* + * 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.rpc; + +import com.google.api.core.BetaApi; + +/** Status snapshot of an ongoing resumable upload. */ +@BetaApi +public final class ResumableUploadStatus { + private final long bytesUploaded; + private final long totalBytes; + private final ResumableUploadProgressListener.State state; + + public ResumableUploadStatus( + long bytesUploaded, long totalBytes, ResumableUploadProgressListener.State state) { + this.bytesUploaded = bytesUploaded; + this.totalBytes = totalBytes; + this.state = state; + } + + /** Returns the number of bytes successfully uploaded to the server so far. */ + public long getBytesUploaded() { + return bytesUploaded; + } + + /** Returns the total size of the stream in bytes, or -1 if unknown. */ + public long getTotalBytes() { + return totalBytes; + } + + /** Returns the current state of the upload session. */ + public ResumableUploadProgressListener.State getState() { + return state; + } + + @Override + public String toString() { + return "ResumableUploadStatus{" + + "bytesUploaded=" + + bytesUploaded + + ", totalBytes=" + + totalBytes + + ", state=" + + state + + '}'; + } +} diff --git a/sdk-platform-java/hermetic_build/library_generation/templates/owlbot.yaml.monorepo.j2 b/sdk-platform-java/hermetic_build/library_generation/templates/owlbot.yaml.monorepo.j2 index d3f29de32a7e..5d7fcf8ff34c 100644 --- a/sdk-platform-java/hermetic_build/library_generation/templates/owlbot.yaml.monorepo.j2 +++ b/sdk-platform-java/hermetic_build/library_generation/templates/owlbot.yaml.monorepo.j2 @@ -33,14 +33,19 @@ deep-copy-regex: - source: "/{{ proto_path }}/(v.*)/.*-java/samples/snippets/generated" dest: "/owl-bot-staging/{{ module_name }}/$1/samples/snippets/generated" {%- else %} +{%- if proto_only %} - source: "/{{ proto_path }}/.*-java/proto-google-.*/src" - dest: "/owl-bot-staging/{{ module_name }}/proto-{{ artifact_id }}/src" + dest: "/owl-bot-staging/{{ module_name }}/{{ unversioned_dir }}/{{ artifact_id }}/src" +{%- else %} +- source: "/{{ proto_path }}/.*-java/proto-google-.*/src" + dest: "/owl-bot-staging/{{ module_name }}/{{ unversioned_dir }}/proto-{{ artifact_id }}/src" - source: "/{{ proto_path }}/.*-java/grpc-google-.*/src" - dest: "/owl-bot-staging/{{ module_name }}/grpc-{{ artifact_id }}/src" + dest: "/owl-bot-staging/{{ module_name }}/{{ unversioned_dir }}/grpc-{{ artifact_id }}/src" - source: "/{{ proto_path }}/.*-java/gapic-google-.*/src" - dest: "/owl-bot-staging/{{ module_name }}/{{ artifact_id }}/src" + dest: "/owl-bot-staging/{{ module_name }}/{{ unversioned_dir }}/{{ artifact_id }}/src" - source: "/{{ proto_path }}/.*-java/samples/snippets/generated" - dest: "/owl-bot-staging/{{ module_name }}/samples/snippets/generated" + dest: "/owl-bot-staging/{{ module_name }}/{{ unversioned_dir }}/samples/snippets/generated" +{%- endif %} {%- endif %} {%- endif %} diff --git a/sdk-platform-java/hermetic_build/library_generation/tests/resources/goldens/.OwlBot-hermetic-unversioned-gapic-golden.yaml b/sdk-platform-java/hermetic_build/library_generation/tests/resources/goldens/.OwlBot-hermetic-unversioned-gapic-golden.yaml new file mode 100644 index 000000000000..33dcf88e4e32 --- /dev/null +++ b/sdk-platform-java/hermetic_build/library_generation/tests/resources/goldens/.OwlBot-hermetic-unversioned-gapic-golden.yaml @@ -0,0 +1,35 @@ +# Copyright 2024 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +deep-remove-regex: +- "/java-bare-metal-solution/grpc-google-.*/src" +- "/java-bare-metal-solution/proto-google-.*/src" +- "/java-bare-metal-solution/google-.*/src" +- "/java-bare-metal-solution/samples/snippets/generated" + +deep-preserve-regex: +- "/java-bare-metal-solution/google-.*/src/test/java/com/google/cloud/.*/v.*/it/IT.*Test.java" + +deep-copy-regex: +- source: "/google/cloud/baremetalsolution/.*-java/proto-google-.*/src" + dest: "/owl-bot-staging/java-bare-metal-solution/baremetalsolution/proto-google-cloud-bare-metal-solution/src" +- source: "/google/cloud/baremetalsolution/.*-java/grpc-google-.*/src" + dest: "/owl-bot-staging/java-bare-metal-solution/baremetalsolution/grpc-google-cloud-bare-metal-solution/src" +- source: "/google/cloud/baremetalsolution/.*-java/gapic-google-.*/src" + dest: "/owl-bot-staging/java-bare-metal-solution/baremetalsolution/google-cloud-bare-metal-solution/src" +- source: "/google/cloud/baremetalsolution/.*-java/samples/snippets/generated" + dest: "/owl-bot-staging/java-bare-metal-solution/baremetalsolution/samples/snippets/generated" + +api-name: baremetalsolution \ No newline at end of file diff --git a/sdk-platform-java/hermetic_build/library_generation/tests/resources/goldens/.OwlBot-hermetic-unversioned-proto-only-golden.yaml b/sdk-platform-java/hermetic_build/library_generation/tests/resources/goldens/.OwlBot-hermetic-unversioned-proto-only-golden.yaml new file mode 100644 index 000000000000..a93303a46d7d --- /dev/null +++ b/sdk-platform-java/hermetic_build/library_generation/tests/resources/goldens/.OwlBot-hermetic-unversioned-proto-only-golden.yaml @@ -0,0 +1,29 @@ +# Copyright 2024 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +deep-remove-regex: +- "/java-bare-metal-solution/grpc-google-.*/src" +- "/java-bare-metal-solution/proto-google-.*/src" +- "/java-bare-metal-solution/google-.*/src" +- "/java-bare-metal-solution/samples/snippets/generated" + +deep-preserve-regex: +- "/java-bare-metal-solution/google-.*/src/test/java/com/google/cloud/.*/v.*/it/IT.*Test.java" + +deep-copy-regex: +- source: "/google/cloud/baremetalsolution/.*-java/proto-google-.*/src" + dest: "/owl-bot-staging/java-bare-metal-solution/baremetalsolution/google-cloud-bare-metal-solution/src" + +api-name: baremetalsolution \ No newline at end of file diff --git a/sdk-platform-java/hermetic_build/library_generation/tests/utilities_unit_tests.py b/sdk-platform-java/hermetic_build/library_generation/tests/utilities_unit_tests.py index 0708fdb4c556..72ccaa89c544 100644 --- a/sdk-platform-java/hermetic_build/library_generation/tests/utilities_unit_tests.py +++ b/sdk-platform-java/hermetic_build/library_generation/tests/utilities_unit_tests.py @@ -253,6 +253,38 @@ def test_generate_postprocessing_prerequisite_files_proto_only_repo_success(self ) self.__remove_postprocessing_prerequisite_files(path=library_path) + def test_generate_postprocessing_prerequisite_files_unversioned_proto_only_success(self): + self.maxDiff = None + library_path = self.__setup_postprocessing_prerequisite_files( + combination=3, + library_type="OTHER", + proto_path="google/cloud/baremetalsolution", + has_version=False, + proto_only=True, + ) + + file_comparator.compare_files( + f"{library_path}/.OwlBot-hermetic.yaml", + f"{library_path}/.OwlBot-hermetic-unversioned-proto-only-golden.yaml", + ) + self.__remove_postprocessing_prerequisite_files(path=library_path) + + def test_generate_postprocessing_prerequisite_files_unversioned_gapic_success(self): + self.maxDiff = None + library_path = self.__setup_postprocessing_prerequisite_files( + combination=2, + library_type="GAPIC_AUTO", + proto_path="google/cloud/baremetalsolution", + has_version=False, + proto_only=False, + ) + + file_comparator.compare_files( + f"{library_path}/.OwlBot-hermetic.yaml", + f"{library_path}/.OwlBot-hermetic-unversioned-gapic-golden.yaml", + ) + self.__remove_postprocessing_prerequisite_files(path=library_path) + def test_generate_postprocessing_prerequisite_files__custom_transport_set_in_config__success( self, ): @@ -326,6 +358,9 @@ def __setup_postprocessing_prerequisite_files( combination: int, library_type: str = "GAPIC_AUTO", library: LibraryConfig = library_1, + proto_path: str = "google/cloud/baremetalsolution/v2", + has_version: bool = True, + proto_only: bool = False, ) -> str: library_path = f"{resources_dir}/goldens" files = [ @@ -336,7 +371,6 @@ def __setup_postprocessing_prerequisite_files( cleanup(files) library.library_type = library_type config = self.__get_a_gen_config(combination, library_type=library_type) - proto_path = "google/cloud/baremetalsolution/v2" gapic_inputs = GapicInputs() # defaults to transport=grpc transport = library.get_transport(gapic_inputs) util.generate_postprocessing_prerequisite_files( @@ -345,6 +379,8 @@ def __setup_postprocessing_prerequisite_files( proto_path=proto_path, transport=transport, library_path=library_path, + has_version=has_version, + proto_only=proto_only, ) return library_path diff --git a/sdk-platform-java/hermetic_build/library_generation/utils/utilities.py b/sdk-platform-java/hermetic_build/library_generation/utils/utilities.py index 23fd92458919..9e9edc00d7b1 100755 --- a/sdk-platform-java/hermetic_build/library_generation/utils/utilities.py +++ b/sdk-platform-java/hermetic_build/library_generation/utils/utilities.py @@ -206,6 +206,7 @@ def generate_postprocessing_prerequisite_files( library_path: str, language: str = "java", has_version: bool = True, + proto_only: bool = False, ) -> None: """ Generates the postprocessing prerequisite files for a library. @@ -301,6 +302,7 @@ def generate_postprocessing_prerequisite_files( else f"{library_path}/.github/{owlbot_yaml_file}" ) if not os.path.exists(path_to_owlbot_yaml_file): + unversioned_dir = remove_version_from(proto_path).split("/")[-1] render( template_name="owlbot.yaml.monorepo.j2", output_name=path_to_owlbot_yaml_file, @@ -309,6 +311,8 @@ def generate_postprocessing_prerequisite_files( module_name=repo_metadata["repo_short"], api_shortname=library.api_shortname, has_version=has_version, + proto_only=proto_only, + unversioned_dir=unversioned_dir, ) # generate owlbot.py