Skip to content

Commit 75003f3

Browse files
whoweswhowes
committed
feat(gax): add progress listener models and UploadProgressTracker
1 parent 17e0368 commit 75003f3

5 files changed

Lines changed: 718 additions & 0 deletions

File tree

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

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
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 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+
/** Returns the exception that triggered recovery or caused failure, if any. */
79+
public abstract @Nullable Throwable getException();
80+
81+
public abstract Builder toBuilder();
82+
83+
public static Builder newBuilder() {
84+
return new AutoValue_ResumableUploadProgress.Builder()
85+
.setBytesUploaded(0L)
86+
.setState(State.STARTING);
87+
}
88+
89+
@AutoValue.Builder
90+
public abstract static class Builder {
91+
public abstract Builder setUploadUrl(@Nullable String uploadUrl);
92+
93+
public abstract Builder setBytesUploaded(long bytesUploaded);
94+
95+
public abstract Builder setState(State state);
96+
97+
public abstract Builder setException(@Nullable Throwable exception);
98+
99+
public abstract ResumableUploadProgress build();
100+
}
101+
}

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
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 com.google.api.core.BetaApi;
33+
import org.jspecify.annotations.NullMarked;
34+
35+
/** A callback listener for observing progress and state transitions of a resumable upload. */
36+
@BetaApi
37+
@FunctionalInterface
38+
@NullMarked
39+
public interface ResumableUploadProgressListener {
40+
41+
/**
42+
* Invoked when upload progress or state changes.
43+
*
44+
* <p>Cancellation via {@link ResumableUploadFuture#cancel(boolean)} can be invoked safely from
45+
* within this callback.
46+
*
47+
* @param progress the current progress snapshot of the upload
48+
*/
49+
void onProgress(ResumableUploadProgress progress);
50+
}

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

Lines changed: 239 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,239 @@
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.common.util.concurrent.MoreExecutors;
35+
import com.google.errorprone.annotations.concurrent.GuardedBy;
36+
import java.util.ArrayList;
37+
import java.util.List;
38+
import java.util.concurrent.Executor;
39+
import org.jspecify.annotations.NullMarked;
40+
import org.jspecify.annotations.Nullable;
41+
42+
/**
43+
* Thread-safe tracker and dispatcher for resumable upload progress and state transitions.
44+
*
45+
* <p>Enforces monotonic progress reporting, isolates listeners from upload pipeline failures,
46+
* serializes callbacks per listener, and manages transition to terminal states.
47+
*/
48+
@NullMarked
49+
class UploadProgressTracker {
50+
51+
private static final class RegisteredListener {
52+
final ResumableUploadProgressListener listener;
53+
final Executor sequentialExecutor;
54+
55+
RegisteredListener(ResumableUploadProgressListener listener, Executor executor) {
56+
this.listener = listener;
57+
this.sequentialExecutor = MoreExecutors.newSequentialExecutor(executor);
58+
}
59+
}
60+
61+
private final Object lock = new Object();
62+
63+
@GuardedBy("lock")
64+
private final List<RegisteredListener> listeners = new ArrayList<>();
65+
66+
@GuardedBy("lock")
67+
private ResumableUploadProgress currentStatus;
68+
69+
@GuardedBy("lock")
70+
private boolean terminal;
71+
72+
@GuardedBy("lock")
73+
private @Nullable String uploadSessionUrl;
74+
75+
UploadProgressTracker() {
76+
this.currentStatus =
77+
ResumableUploadProgress.newBuilder()
78+
.setState(ResumableUploadProgress.State.STARTING)
79+
.setBytesUploaded(0L)
80+
.build();
81+
}
82+
83+
void addListener(ResumableUploadProgressListener listener, Executor executor) {
84+
checkNotNull(listener, "listener must not be null");
85+
checkNotNull(executor, "executor must not be null");
86+
RegisteredListener entry = new RegisteredListener(listener, executor);
87+
ResumableUploadProgress snapshot;
88+
synchronized (lock) {
89+
snapshot = this.currentStatus;
90+
if (!terminal) {
91+
listeners.add(entry);
92+
}
93+
}
94+
entry.sequentialExecutor.execute(() -> dispatchSafely(listener, snapshot));
95+
}
96+
97+
ResumableUploadProgress getStatus() {
98+
synchronized (lock) {
99+
return currentStatus;
100+
}
101+
}
102+
103+
void onStarted(String uploadUrl) {
104+
checkNotNull(uploadUrl, "uploadUrl must not be null");
105+
List<RegisteredListener> snapshot;
106+
ResumableUploadProgress status;
107+
synchronized (lock) {
108+
this.uploadSessionUrl = uploadUrl;
109+
status =
110+
currentStatus.toBuilder()
111+
.setState(ResumableUploadProgress.State.STARTED)
112+
.setUploadUrl(uploadUrl)
113+
.build();
114+
snapshot = updateStatusLocked(status);
115+
}
116+
notifyListeners(snapshot, status);
117+
}
118+
119+
void onChunkUploaded(long bytesUploaded) {
120+
List<RegisteredListener> snapshot;
121+
ResumableUploadProgress status;
122+
synchronized (lock) {
123+
if (terminal) {
124+
return;
125+
}
126+
long bytes = Math.max(currentStatus.getBytesUploaded(), bytesUploaded);
127+
status =
128+
currentStatus.toBuilder()
129+
.setState(ResumableUploadProgress.State.UPLOADING)
130+
.setBytesUploaded(bytes)
131+
.setUploadUrl(uploadSessionUrl)
132+
.build();
133+
snapshot = updateStatusLocked(status);
134+
}
135+
notifyListeners(snapshot, status);
136+
}
137+
138+
void onRecovering(@Nullable Throwable cause) {
139+
List<RegisteredListener> snapshot;
140+
ResumableUploadProgress status;
141+
synchronized (lock) {
142+
if (terminal) {
143+
return;
144+
}
145+
status =
146+
currentStatus.toBuilder()
147+
.setState(ResumableUploadProgress.State.RECOVERING)
148+
.setException(cause)
149+
.setUploadUrl(uploadSessionUrl)
150+
.build();
151+
snapshot = updateStatusLocked(status);
152+
}
153+
notifyListeners(snapshot, status);
154+
}
155+
156+
void onOffsetReceived(long committedOffset) {
157+
List<RegisteredListener> snapshot;
158+
ResumableUploadProgress status;
159+
synchronized (lock) {
160+
if (terminal) {
161+
return;
162+
}
163+
long bytes = Math.max(currentStatus.getBytesUploaded(), committedOffset);
164+
status =
165+
currentStatus.toBuilder()
166+
.setState(ResumableUploadProgress.State.OFFSET_RECEIVED)
167+
.setBytesUploaded(bytes)
168+
.setUploadUrl(uploadSessionUrl)
169+
.build();
170+
snapshot = updateStatusLocked(status);
171+
}
172+
notifyListeners(snapshot, status);
173+
}
174+
175+
void onFinalized(long totalBytes) {
176+
List<RegisteredListener> snapshot;
177+
ResumableUploadProgress status;
178+
synchronized (lock) {
179+
if (terminal) {
180+
return;
181+
}
182+
terminal = true;
183+
long bytes = Math.max(currentStatus.getBytesUploaded(), totalBytes);
184+
status =
185+
currentStatus.toBuilder()
186+
.setState(ResumableUploadProgress.State.FINALIZED)
187+
.setBytesUploaded(bytes)
188+
.setUploadUrl(uploadSessionUrl)
189+
.build();
190+
snapshot = updateStatusLocked(status);
191+
}
192+
notifyListeners(snapshot, status);
193+
}
194+
195+
void onFailed(@Nullable Throwable error, @Nullable String sessionUrl) {
196+
List<RegisteredListener> snapshot;
197+
ResumableUploadProgress status;
198+
synchronized (lock) {
199+
if (terminal) {
200+
return;
201+
}
202+
terminal = true;
203+
String url = sessionUrl != null ? sessionUrl : uploadSessionUrl;
204+
status =
205+
currentStatus.toBuilder()
206+
.setState(ResumableUploadProgress.State.FAILED)
207+
.setException(error)
208+
.setUploadUrl(url)
209+
.build();
210+
snapshot = updateStatusLocked(status);
211+
}
212+
notifyListeners(snapshot, status);
213+
}
214+
215+
@GuardedBy("lock")
216+
private List<RegisteredListener> updateStatusLocked(ResumableUploadProgress newStatus) {
217+
this.currentStatus = newStatus;
218+
if (newStatus.getUploadUrl() != null && this.uploadSessionUrl == null) {
219+
this.uploadSessionUrl = newStatus.getUploadUrl();
220+
}
221+
return new ArrayList<>(this.listeners);
222+
}
223+
224+
private void notifyListeners(
225+
List<RegisteredListener> targetListeners, ResumableUploadProgress status) {
226+
for (RegisteredListener entry : targetListeners) {
227+
entry.sequentialExecutor.execute(() -> dispatchSafely(entry.listener, status));
228+
}
229+
}
230+
231+
private static void dispatchSafely(
232+
ResumableUploadProgressListener listener, ResumableUploadProgress status) {
233+
try {
234+
listener.onProgress(status);
235+
} catch (Throwable ignored) {
236+
// Listener exceptions are isolated from the upload pipeline
237+
}
238+
}
239+
}

0 commit comments

Comments
 (0)