Skip to content

Commit 391a7c4

Browse files
whoweswhowes
committed
feat(gax): add progress listener models and UploadProgressTracker
1 parent 9b5543b commit 391a7c4

5 files changed

Lines changed: 725 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: 246 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,246 @@
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+
if (terminal) {
109+
return;
110+
}
111+
this.uploadSessionUrl = uploadUrl;
112+
status =
113+
currentStatus.toBuilder()
114+
.setState(ResumableUploadProgress.State.STARTED)
115+
.setUploadUrl(uploadUrl)
116+
.build();
117+
snapshot = updateStatusLocked(status);
118+
}
119+
notifyListeners(snapshot, status);
120+
}
121+
122+
void onChunkUploaded(long bytesUploaded) {
123+
List<RegisteredListener> snapshot;
124+
ResumableUploadProgress status;
125+
synchronized (lock) {
126+
if (terminal) {
127+
return;
128+
}
129+
long bytes = Math.max(currentStatus.getBytesUploaded(), bytesUploaded);
130+
status =
131+
currentStatus.toBuilder()
132+
.setState(ResumableUploadProgress.State.UPLOADING)
133+
.setBytesUploaded(bytes)
134+
.setUploadUrl(uploadSessionUrl)
135+
.build();
136+
snapshot = updateStatusLocked(status);
137+
}
138+
notifyListeners(snapshot, status);
139+
}
140+
141+
void onRecovering(@Nullable Throwable cause) {
142+
List<RegisteredListener> snapshot;
143+
ResumableUploadProgress status;
144+
synchronized (lock) {
145+
if (terminal) {
146+
return;
147+
}
148+
status =
149+
currentStatus.toBuilder()
150+
.setState(ResumableUploadProgress.State.RECOVERING)
151+
.setException(cause)
152+
.setUploadUrl(uploadSessionUrl)
153+
.build();
154+
snapshot = updateStatusLocked(status);
155+
}
156+
notifyListeners(snapshot, status);
157+
}
158+
159+
void onOffsetReceived(long committedOffset) {
160+
List<RegisteredListener> snapshot;
161+
ResumableUploadProgress status;
162+
synchronized (lock) {
163+
if (terminal) {
164+
return;
165+
}
166+
long bytes = Math.max(currentStatus.getBytesUploaded(), committedOffset);
167+
status =
168+
currentStatus.toBuilder()
169+
.setState(ResumableUploadProgress.State.OFFSET_RECEIVED)
170+
.setBytesUploaded(bytes)
171+
.setUploadUrl(uploadSessionUrl)
172+
.build();
173+
snapshot = updateStatusLocked(status);
174+
}
175+
notifyListeners(snapshot, status);
176+
}
177+
178+
void onFinalized(long totalBytes) {
179+
List<RegisteredListener> snapshot;
180+
ResumableUploadProgress status;
181+
synchronized (lock) {
182+
if (terminal) {
183+
return;
184+
}
185+
terminal = true;
186+
long bytes = Math.max(currentStatus.getBytesUploaded(), totalBytes);
187+
status =
188+
currentStatus.toBuilder()
189+
.setState(ResumableUploadProgress.State.FINALIZED)
190+
.setBytesUploaded(bytes)
191+
.setUploadUrl(uploadSessionUrl)
192+
.build();
193+
snapshot = updateStatusLocked(status);
194+
}
195+
notifyListeners(snapshot, status);
196+
}
197+
198+
void onFailed(@Nullable Throwable error, @Nullable String sessionUrl) {
199+
List<RegisteredListener> snapshot;
200+
ResumableUploadProgress status;
201+
synchronized (lock) {
202+
if (terminal) {
203+
return;
204+
}
205+
terminal = true;
206+
String url = sessionUrl != null ? sessionUrl : uploadSessionUrl;
207+
status =
208+
currentStatus.toBuilder()
209+
.setState(ResumableUploadProgress.State.FAILED)
210+
.setException(error)
211+
.setUploadUrl(url)
212+
.build();
213+
snapshot = updateStatusLocked(status);
214+
}
215+
notifyListeners(snapshot, status);
216+
}
217+
218+
@GuardedBy("lock")
219+
private List<RegisteredListener> updateStatusLocked(ResumableUploadProgress newStatus) {
220+
this.currentStatus = newStatus;
221+
if (newStatus.getUploadUrl() != null && this.uploadSessionUrl == null) {
222+
this.uploadSessionUrl = newStatus.getUploadUrl();
223+
}
224+
List<RegisteredListener> snapshot = new ArrayList<>(this.listeners);
225+
if (this.terminal) {
226+
this.listeners.clear();
227+
}
228+
return snapshot;
229+
}
230+
231+
private void notifyListeners(
232+
List<RegisteredListener> targetListeners, ResumableUploadProgress status) {
233+
for (RegisteredListener entry : targetListeners) {
234+
entry.sequentialExecutor.execute(() -> dispatchSafely(entry.listener, status));
235+
}
236+
}
237+
238+
private static void dispatchSafely(
239+
ResumableUploadProgressListener listener, ResumableUploadProgress status) {
240+
try {
241+
listener.onProgress(status);
242+
} catch (Throwable ignored) {
243+
// Listener exceptions are isolated from the upload pipeline
244+
}
245+
}
246+
}

0 commit comments

Comments
 (0)