feat(gax): add ResumableUploadProgressListener and related infrastructure

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();
+  }
+}

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);
+}

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<ResumableUploadProgress> 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<RegisteredListener> 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<ResumableUploadProgress.Builder> statusUpdater) {
+    List<RegisteredListener> 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
+    }
+  }
+}