From 15c09c7d9871becb28422d52072d3af207c1b59a Mon Sep 17 00:00:00 2001 From: whowes Date: Sat, 12 Sep 2026 17:07:50 +0000 Subject: [PATCH] feat(gax): add progress listener models and ResumableUploadProgressTracker --- .../resumable/ResumableUploadProgress.java | 98 ++++++++++ .../ResumableUploadProgressListener.java | 47 +++++ .../rpc/ResumableUploadProgressTracker.java | 179 ++++++++++++++++++ .../ResumableUploadProgressTrackerTest.java | 179 ++++++++++++++++++ 4 files changed, 503 insertions(+) create mode 100644 sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgress.java create mode 100644 sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java create mode 100644 sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java create mode 100644 sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadProgressTrackerTest.java diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgress.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgress.java new file mode 100644 index 000000000000..0b4bf5c115be --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgress.java @@ -0,0 +1,98 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** Progress snapshot of an ongoing or completed resumable upload session. */ +@BetaApi +@NullMarked +@AutoValue +public abstract class ResumableUploadProgress { + + ResumableUploadProgress() {} + + /** The state of the resumable upload session. */ + public enum State { + /** Session initiation is in progress (acquiring upload session URL). */ + STARTING, + + /** The session initiation completed successfully. */ + STARTED, + + /** Transmitting chunk payloads to the server. */ + UPLOADING, + + /** A recoverable error occurred; querying server status and resynchronizing offset. */ + RECOVERING, + + /** The server query status succeeded and the committed offset was received. */ + OFFSET_RECEIVED, + + /** The upload was successfully finalized by the server. */ + FINALIZED, + + /** The upload failed unrecoverably or was cancelled. */ + FAILED + } + + /** + * Returns the negotiated upload session URI, or {@code null} if session initiation is pending. + */ + public abstract @Nullable String getUploadUrl(); + + /** Returns the number of bytes confirmed as uploaded to the server so far. */ + public abstract long getBytesUploaded(); + + /** Returns the current state of the upload session. */ + public abstract State getState(); + + public abstract Builder toBuilder(); + + public static Builder newBuilder() { + return new AutoValue_ResumableUploadProgress.Builder() + .setBytesUploaded(0L) + .setState(State.STARTING); + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setUploadUrl(@Nullable String uploadUrl); + + public abstract Builder setBytesUploaded(long bytesUploaded); + + public abstract Builder setState(State state); + + public abstract ResumableUploadProgress build(); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java new file mode 100644 index 000000000000..b46e5cc648d8 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java @@ -0,0 +1,47 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import com.google.api.core.BetaApi; +import org.jspecify.annotations.NullMarked; + +/** A callback listener for observing progress and state transitions of a resumable upload. */ +@BetaApi +@FunctionalInterface +@NullMarked +public interface ResumableUploadProgressListener { + + /** + * Invoked when upload progress or state changes. + * + * @param progress the current progress snapshot of the upload + */ + void onProgress(ResumableUploadProgress progress); +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java new file mode 100644 index 000000000000..102654dcdbf3 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java @@ -0,0 +1,179 @@ +/* + * 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 static com.google.common.base.Preconditions.checkNotNull; + +import com.google.api.gax.resumable.ResumableUploadProgress; +import com.google.api.gax.resumable.ResumableUploadProgressListener; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.errorprone.annotations.concurrent.GuardedBy; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import java.util.function.Consumer; +import org.jspecify.annotations.NullMarked; + +/** Publishes updates to listeners on resumable upload progress and state transitions. */ +@NullMarked +class ResumableUploadProgressTracker { + + private static final class RegisteredListener { + final ResumableUploadProgressListener listener; + final Executor sequentialExecutor; + final ConcurrentLinkedQueue pendingUpdates = + new ConcurrentLinkedQueue<>(); + + RegisteredListener(ResumableUploadProgressListener listener, Executor executor) { + this.listener = listener; + this.sequentialExecutor = MoreExecutors.newSequentialExecutor(executor); + } + + void drain() { + try { + sequentialExecutor.execute( + () -> { + ResumableUploadProgress status; + while ((status = pendingUpdates.poll()) != null) { + dispatchSafely(listener, status); + } + }); + } catch (RejectedExecutionException ignored) { + // Executor exceptions are isolated from the upload pipeline + } + } + } + + private final Object lock = new Object(); + + @GuardedBy("lock") + private final List listeners = new ArrayList<>(); + + @GuardedBy("lock") + private ResumableUploadProgress currentStatus; + + ResumableUploadProgressTracker() { + this.currentStatus = + ResumableUploadProgress.newBuilder() + .setState(ResumableUploadProgress.State.STARTING) + .setBytesUploaded(0L) + .build(); + } + + void addListener(ResumableUploadProgressListener listener, Executor executor) { + checkNotNull(listener, "listener must not be null"); + checkNotNull(executor, "executor must not be null"); + RegisteredListener entry = new RegisteredListener(listener, executor); + synchronized (lock) { + entry.pendingUpdates.add(this.currentStatus); + if (!isTerminal()) { + listeners.add(entry); + } + } + entry.drain(); + } + + ResumableUploadProgress getStatus() { + synchronized (lock) { + return currentStatus; + } + } + + void onStarted(String uploadUrl) { + checkNotNull(uploadUrl, "uploadUrl must not be null"); + transition(b -> b.setState(ResumableUploadProgress.State.STARTED).setUploadUrl(uploadUrl)); + } + + void onChunkUploaded(long bytesUploaded) { + transition( + b -> b.setState(ResumableUploadProgress.State.UPLOADING).setBytesUploaded(bytesUploaded)); + } + + void onRecovering() { + transition(b -> b.setState(ResumableUploadProgress.State.RECOVERING)); + } + + void onOffsetReceived(long committedOffset) { + transition( + b -> + b.setState(ResumableUploadProgress.State.OFFSET_RECEIVED) + .setBytesUploaded(committedOffset)); + } + + void onFinalized(long totalBytes) { + transition( + b -> b.setState(ResumableUploadProgress.State.FINALIZED).setBytesUploaded(totalBytes)); + } + + void onFailed() { + transition(b -> b.setState(ResumableUploadProgress.State.FAILED)); + } + + @GuardedBy("lock") + private boolean isTerminal() { + ResumableUploadProgress.State state = currentStatus.getState(); + return state == ResumableUploadProgress.State.FINALIZED + || state == ResumableUploadProgress.State.FAILED; + } + + private void transition(Consumer statusUpdater) { + List snapshot; + synchronized (lock) { + if (isTerminal()) { + return; + } + ResumableUploadProgress.Builder builder = currentStatus.toBuilder(); + statusUpdater.accept(builder); + ResumableUploadProgress newStatus = builder.build(); + this.currentStatus = newStatus; + for (RegisteredListener entry : this.listeners) { + entry.pendingUpdates.add(newStatus); + } + snapshot = new ArrayList<>(this.listeners); + if (isTerminal()) { + this.listeners.clear(); + } + } + for (RegisteredListener entry : snapshot) { + entry.drain(); + } + } + + private static void dispatchSafely( + ResumableUploadProgressListener listener, ResumableUploadProgress status) { + try { + listener.onProgress(status); + } catch (Throwable ignored) { + // Listener exceptions are isolated from the upload pipeline + } + } +} diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadProgressTrackerTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadProgressTrackerTest.java new file mode 100644 index 000000000000..e33010b9df70 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadProgressTrackerTest.java @@ -0,0 +1,179 @@ +/* + * 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 static com.google.common.truth.Truth.assertThat; + +import com.google.api.gax.resumable.ResumableUploadProgress; +import com.google.common.util.concurrent.MoreExecutors; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +class ResumableUploadProgressTrackerTest { + + @Test + void testInitialSnapshotOnSubscribe() { + ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + List statuses = new ArrayList<>(); + tracker.addListener(statuses::add, MoreExecutors.directExecutor()); + + assertThat(statuses).hasSize(1); + ResumableUploadProgress initial = statuses.get(0); + assertThat(initial.getState()).isEqualTo(ResumableUploadProgress.State.STARTING); + assertThat(initial.getBytesUploaded()).isEqualTo(0L); + assertThat(initial.getUploadUrl()).isNull(); + } + + @Test + void testProgressUpdates() { + ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + tracker.onStarted("https://upload.url/test"); + + List byteUpdates = new ArrayList<>(); + tracker.addListener( + status -> byteUpdates.add(status.getBytesUploaded()), MoreExecutors.directExecutor()); + + // Advance to 100 bytes + tracker.onChunkUploaded(100L); + assertThat(byteUpdates).containsExactly(0L, 100L).inOrder(); + + // Advance to 200 bytes + tracker.onChunkUploaded(200L); + assertThat(byteUpdates).containsExactly(0L, 100L, 200L).inOrder(); + } + + @Test + void testListenerExceptionSafety_doesNotDisruptSubsequentUpdates() { + ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + tracker.onStarted("https://upload.url/test"); + + AtomicInteger errorCount = new AtomicInteger(0); + List safeReceived = new ArrayList<>(); + + // Listener 1 throws on every call + tracker.addListener( + status -> { + errorCount.incrementAndGet(); + throw new RuntimeException("boom from listener 1"); + }, + MoreExecutors.directExecutor()); + + // Listener 2 functions normally + tracker.addListener( + status -> safeReceived.add(status.getState()), MoreExecutors.directExecutor()); + + tracker.onChunkUploaded(50L); + tracker.onFinalized(50L); + + assertThat(errorCount.get()).isEqualTo(3); + assertThat(safeReceived) + .containsExactly( + ResumableUploadProgress.State.STARTED, + ResumableUploadProgress.State.UPLOADING, + ResumableUploadProgress.State.FINALIZED) + .inOrder(); + } + + @Test + void testTerminalState_noNotificationsAfterFinalized() { + ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + tracker.onStarted("https://upload.url/test"); + + List received = new ArrayList<>(); + tracker.addListener(status -> received.add(status.getState()), MoreExecutors.directExecutor()); + + tracker.onChunkUploaded(100L); + tracker.onFinalized(100L); + + // Updates after finalized must be discarded + tracker.onChunkUploaded(150L); + tracker.onRecovering(); + + assertThat(received) + .containsExactly( + ResumableUploadProgress.State.STARTED, + ResumableUploadProgress.State.UPLOADING, + ResumableUploadProgress.State.FINALIZED) + .inOrder(); + assertThat(tracker.getStatus().getUploadUrl()).isEqualTo("https://upload.url/test"); + } + + @Test + void testListenerReentrancy_canQueryStatusInsideCallback() { + ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + AtomicReference observed = new AtomicReference<>(); + + tracker.addListener( + status -> observed.set(tracker.getStatus()), MoreExecutors.directExecutor()); + + tracker.onStarted("https://upload.url/reentrancy"); + assertThat(observed.get()).isNotNull(); + assertThat(observed.get().getState()).isEqualTo(ResumableUploadProgress.State.STARTED); + assertThat(observed.get().getUploadUrl()).isEqualTo("https://upload.url/reentrancy"); + } + + @Test + void testCustomExecutorDispatch() throws Exception { + ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + CountDownLatch latch = new CountDownLatch(2); + List states = new ArrayList<>(); + + tracker.addListener( + status -> { + synchronized (states) { + states.add(status.getState()); + } + latch.countDown(); + }, + executor); + + tracker.onStarted("https://upload.url/custom-executor"); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + + synchronized (states) { + assertThat(states) + .containsExactly( + ResumableUploadProgress.State.STARTING, ResumableUploadProgress.State.STARTED) + .inOrder(); + } + } finally { + executor.shutdownNow(); + } + } +}