Skip to content

Commit 161d165

Browse files
whoweswhowes
committed
feat(gax): add progress listener models and ResumableUploadProgressTracker
1 parent 1c271f8 commit 161d165

5 files changed

Lines changed: 661 additions & 0 deletions

File tree

‎sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgress.java‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Redistribution and use in source and binary forms, with or without
5+
* modification, are permitted provided that the following conditions are
6+
* met:
7+
*
8+
* * Redistributions of source code must retain the above copyright
9+
* notice, this list of conditions and the following disclaimer.
10+
* * Redistributions in binary form must reproduce the above
11+
* copyright notice, this list of conditions and the following disclaimer
12+
* in the documentation and/or other materials provided with the
13+
* distribution.
14+
* * Neither the name of Google LLC nor the names of its
15+
* contributors may be used to endorse or promote products derived from
16+
* this software without specific prior written permission.
17+
*
18+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
21+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
24+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
25+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
26+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
28+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29+
*/
30+
package com.google.api.gax.resumable;
31+
32+
import com.google.api.core.BetaApi;
33+
import com.google.auto.value.AutoValue;
34+
import org.jspecify.annotations.NullMarked;
35+
import org.jspecify.annotations.Nullable;
36+
37+
/** Progress snapshot of an ongoing or completed resumable upload session. */
38+
@BetaApi
39+
@NullMarked
40+
@AutoValue
41+
public abstract class ResumableUploadProgress {
42+
43+
/** The state of the resumable upload session. */
44+
public enum State {
45+
/** Session initiation is in progress (acquiring upload session URL). */
46+
STARTING,
47+
48+
/** The session initiation completed successfully. */
49+
STARTED,
50+
51+
/** Transmitting chunk payloads to the server. */
52+
UPLOADING,
53+
54+
/** A recoverable error occurred; querying server status and resynchronizing offset. */
55+
RECOVERING,
56+
57+
/** The server query status succeeded and the committed offset was received. */
58+
OFFSET_RECEIVED,
59+
60+
/** The upload was successfully finalized by the server. */
61+
FINALIZED,
62+
63+
/** The upload failed unrecoverably or was cancelled. */
64+
FAILED
65+
}
66+
67+
/**
68+
* Returns the negotiated upload session URI, or {@code null} if session initiation is pending.
69+
*/
70+
public abstract @Nullable String getUploadUrl();
71+
72+
/** Returns the number of bytes confirmed as uploaded to the server so far. */
73+
public abstract long getBytesUploaded();
74+
75+
/** Returns the current state of the upload session. */
76+
public abstract State getState();
77+
78+
public abstract Builder toBuilder();
79+
80+
public static Builder newBuilder() {
81+
return new AutoValue_ResumableUploadProgress.Builder()
82+
.setBytesUploaded(0L)
83+
.setState(State.STARTING);
84+
}
85+
86+
@AutoValue.Builder
87+
public abstract static class Builder {
88+
public abstract Builder setUploadUrl(@Nullable String uploadUrl);
89+
90+
public abstract Builder setBytesUploaded(long bytesUploaded);
91+
92+
public abstract Builder setState(State state);
93+
94+
public abstract ResumableUploadProgress build();
95+
}
96+
}

‎sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadProgressListener.java‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Redistribution and use in source and binary forms, with or without
5+
* modification, are permitted provided that the following conditions are
6+
* met:
7+
*
8+
* * Redistributions of source code must retain the above copyright
9+
* notice, this list of conditions and the following disclaimer.
10+
* * Redistributions in binary form must reproduce the above
11+
* copyright notice, this list of conditions and the following disclaimer
12+
* in the documentation and/or other materials provided with the
13+
* distribution.
14+
* * Neither the name of Google LLC nor the names of its
15+
* contributors may be used to endorse or promote products derived from
16+
* this software without specific prior written permission.
17+
*
18+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
21+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
24+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
25+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
26+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
28+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29+
*/
30+
package com.google.api.gax.resumable;
31+
32+
import com.google.api.core.BetaApi;
33+
import com.google.api.gax.rpc.ResumableUploadFuture;
34+
import org.jspecify.annotations.NullMarked;
35+
36+
/** A callback listener for observing progress and state transitions of a resumable upload. */
37+
@BetaApi
38+
@FunctionalInterface
39+
@NullMarked
40+
public interface ResumableUploadProgressListener {
41+
42+
/**
43+
* Invoked when upload progress or state changes.
44+
*
45+
* <p>Cancellation via {@link ResumableUploadFuture#cancel(boolean)} can be invoked safely from
46+
* within this callback.
47+
*
48+
* @param progress the current progress snapshot of the upload
49+
*/
50+
void onProgress(ResumableUploadProgress progress);
51+
}

‎sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java‎

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,177 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Redistribution and use in source and binary forms, with or without
5+
* modification, are permitted provided that the following conditions are
6+
* met:
7+
*
8+
* * Redistributions of source code must retain the above copyright
9+
* notice, this list of conditions and the following disclaimer.
10+
* * Redistributions in binary form must reproduce the above
11+
* copyright notice, this list of conditions and the following disclaimer
12+
* in the documentation and/or other materials provided with the
13+
* distribution.
14+
* * Neither the name of Google LLC nor the names of its
15+
* contributors may be used to endorse or promote products derived from
16+
* this software without specific prior written permission.
17+
*
18+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
21+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
24+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
25+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
26+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
28+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29+
*/
30+
package com.google.api.gax.rpc;
31+
32+
import static com.google.common.base.Preconditions.checkNotNull;
33+
34+
import com.google.api.gax.resumable.ResumableUploadProgress;
35+
import com.google.api.gax.resumable.ResumableUploadProgressListener;
36+
import com.google.common.util.concurrent.MoreExecutors;
37+
import com.google.errorprone.annotations.concurrent.GuardedBy;
38+
import java.util.ArrayList;
39+
import java.util.List;
40+
import java.util.concurrent.Executor;
41+
import java.util.function.Consumer;
42+
import org.jspecify.annotations.NullMarked;
43+
44+
/**
45+
* Thread-safe tracker and dispatcher for resumable upload progress and state transitions.
46+
*
47+
* <p>Isolates listeners from upload pipeline failures, serializes callbacks per listener, and
48+
* manages transition to terminal states.
49+
*/
50+
@NullMarked
51+
class ResumableUploadProgressTracker {
52+
53+
private static final class RegisteredListener {
54+
final ResumableUploadProgressListener listener;
55+
final Executor sequentialExecutor;
56+
57+
RegisteredListener(ResumableUploadProgressListener listener, Executor executor) {
58+
this.listener = listener;
59+
this.sequentialExecutor = MoreExecutors.newSequentialExecutor(executor);
60+
}
61+
}
62+
63+
private final Object lock = new Object();
64+
65+
@GuardedBy("lock")
66+
private final List<RegisteredListener> listeners = new ArrayList<>();
67+
68+
@GuardedBy("lock")
69+
private ResumableUploadProgress currentStatus;
70+
71+
ResumableUploadProgressTracker() {
72+
this.currentStatus =
73+
ResumableUploadProgress.newBuilder()
74+
.setState(ResumableUploadProgress.State.STARTING)
75+
.setBytesUploaded(0L)
76+
.build();
77+
}
78+
79+
void addListener(ResumableUploadProgressListener listener, Executor executor) {
80+
checkNotNull(listener, "listener must not be null");
81+
checkNotNull(executor, "executor must not be null");
82+
RegisteredListener entry = new RegisteredListener(listener, executor);
83+
ResumableUploadProgress snapshot;
84+
synchronized (lock) {
85+
snapshot = this.currentStatus;
86+
if (!isTerminal()) {
87+
listeners.add(entry);
88+
}
89+
}
90+
entry.sequentialExecutor.execute(() -> dispatchSafely(listener, snapshot));
91+
}
92+
93+
ResumableUploadProgress getStatus() {
94+
synchronized (lock) {
95+
return currentStatus;
96+
}
97+
}
98+
99+
void onStarted(String uploadUrl) {
100+
checkNotNull(uploadUrl, "uploadUrl must not be null");
101+
transition(
102+
b ->
103+
b.setState(ResumableUploadProgress.State.STARTED)
104+
.setUploadUrl(uploadUrl));
105+
}
106+
107+
void onChunkUploaded(long bytesUploaded) {
108+
transition(
109+
b ->
110+
b.setState(ResumableUploadProgress.State.UPLOADING)
111+
.setBytesUploaded(bytesUploaded));
112+
}
113+
114+
void onRecovering() {
115+
transition(b -> b.setState(ResumableUploadProgress.State.RECOVERING));
116+
}
117+
118+
void onOffsetReceived(long committedOffset) {
119+
transition(
120+
b ->
121+
b.setState(ResumableUploadProgress.State.OFFSET_RECEIVED)
122+
.setBytesUploaded(committedOffset));
123+
}
124+
125+
void onFinalized(long totalBytes) {
126+
transition(
127+
b ->
128+
b.setState(ResumableUploadProgress.State.FINALIZED)
129+
.setBytesUploaded(totalBytes));
130+
}
131+
132+
void onFailed() {
133+
transition(b -> b.setState(ResumableUploadProgress.State.FAILED));
134+
}
135+
136+
@GuardedBy("lock")
137+
private boolean isTerminal() {
138+
ResumableUploadProgress.State state = currentStatus.getState();
139+
return state == ResumableUploadProgress.State.FINALIZED
140+
|| state == ResumableUploadProgress.State.FAILED;
141+
}
142+
143+
private void transition(Consumer<ResumableUploadProgress.Builder> statusUpdater) {
144+
List<RegisteredListener> snapshot;
145+
ResumableUploadProgress newStatus;
146+
synchronized (lock) {
147+
if (isTerminal()) {
148+
return;
149+
}
150+
ResumableUploadProgress.Builder builder = currentStatus.toBuilder();
151+
statusUpdater.accept(builder);
152+
newStatus = builder.build();
153+
this.currentStatus = newStatus;
154+
snapshot = new ArrayList<>(this.listeners);
155+
if (isTerminal()) {
156+
this.listeners.clear();
157+
}
158+
}
159+
notifyListeners(snapshot, newStatus);
160+
}
161+
162+
private void notifyListeners(
163+
List<RegisteredListener> targetListeners, ResumableUploadProgress status) {
164+
for (RegisteredListener entry : targetListeners) {
165+
entry.sequentialExecutor.execute(() -> dispatchSafely(entry.listener, status));
166+
}
167+
}
168+
169+
private static void dispatchSafely(
170+
ResumableUploadProgressListener listener, ResumableUploadProgress status) {
171+
try {
172+
listener.onProgress(status);
173+
} catch (Throwable ignored) {
174+
// Listener exceptions are isolated from the upload pipeline
175+
}
176+
}
177+
}

0 commit comments

Comments
 (0)