Skip to content

Commit 18db19d

Browse files
whoweswhowes
committed
feat(gax): add progress listener models and UploadProgressTracker
1 parent 4dfdd22 commit 18db19d

5 files changed

Lines changed: 721 additions & 0 deletions

File tree

‎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 status the current status snapshot of the upload
48+
*/
49+
void onProgress(ResumableUploadStatus status);
50+
}

‎sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.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+
/** Status snapshot of an ongoing or completed resumable upload session. */
38+
@BetaApi
39+
@NullMarked
40+
@AutoValue
41+
public abstract class ResumableUploadStatus {
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_ResumableUploadStatus.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 ResumableUploadStatus build();
100+
}
101+
}

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

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

0 commit comments

Comments
 (0)