Skip to content

Commit eec4310

Browse files
blakeli0blakeli0
committed
feat(gax): add ResumableUploadFutureImpl state machine and ResumableUploadClient SPI
- Add ResumableUploadFutureImpl stateful per-request state machine for session initiation and chunk uploads. - Add ResumableUploadClient transport SPI interface (startUploadCallable, uploadChunkCallable) and value objects (ResumableUploadSession, ChunkUploadRequest, ChunkUploadResponse). - Update ResumableUploadCallable to instantiate and start ResumableUploadFutureImpl.
1 parent 04f55d0 commit eec4310

5 files changed

Lines changed: 412 additions & 0 deletions

File tree

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

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
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.common.base.Preconditions;
34+
35+
/** Value object representing a chunk upload request. */
36+
@BetaApi
37+
public class ChunkUploadRequest {
38+
39+
private final String uploadUrl;
40+
private final byte[] data;
41+
private final long offset;
42+
private final long totalSize;
43+
private final boolean isLast;
44+
45+
public ChunkUploadRequest(
46+
String uploadUrl, byte[] data, long offset, long totalSize, boolean isLast) {
47+
this.uploadUrl = Preconditions.checkNotNull(uploadUrl);
48+
this.data = Preconditions.checkNotNull(data);
49+
this.offset = offset;
50+
this.totalSize = totalSize;
51+
this.isLast = isLast;
52+
}
53+
54+
public String getUploadUrl() {
55+
return uploadUrl;
56+
}
57+
58+
public byte[] getData() {
59+
return data;
60+
}
61+
62+
public long getOffset() {
63+
return offset;
64+
}
65+
66+
public long getTotalSize() {
67+
return totalSize;
68+
}
69+
70+
public boolean isLast() {
71+
return isLast;
72+
}
73+
}

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

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
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 javax.annotation.Nullable;
34+
35+
/** Value object representing a chunk upload response. */
36+
@BetaApi
37+
public class ChunkUploadResponse {
38+
39+
private final long committedOffset;
40+
@Nullable private final Object responseBody;
41+
42+
public ChunkUploadResponse(long committedOffset, @Nullable Object responseBody) {
43+
this.committedOffset = committedOffset;
44+
this.responseBody = responseBody;
45+
}
46+
47+
public long getCommittedOffset() {
48+
return committedOffset;
49+
}
50+
51+
@Nullable
52+
public Object getResponseBody() {
53+
return responseBody;
54+
}
55+
}

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

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
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+
34+
/**
35+
* Low-level transport Service Provider Interface (SPI) for executing individual Scotty resumable
36+
* upload operations (session initiation and chunk upload).
37+
*/
38+
@BetaApi
39+
public interface ResumableUploadClient {
40+
41+
/** Returns a UnaryCallable to initiate a new resumable upload session. */
42+
<RequestT> UnaryCallable<RequestT, ResumableUploadSession> startUploadCallable();
43+
44+
/** Returns a UnaryCallable to upload a payload byte chunk to an active session URL. */
45+
UnaryCallable<ChunkUploadRequest, ChunkUploadResponse> uploadChunkCallable();
46+
}

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

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,190 @@
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.AbstractApiFuture;
33+
import com.google.api.core.ApiFuture;
34+
import com.google.api.core.ApiFutureCallback;
35+
import com.google.api.core.ApiFutures;
36+
import com.google.api.core.BetaApi;
37+
import com.google.common.base.Preconditions;
38+
import com.google.common.util.concurrent.MoreExecutors;
39+
import java.io.InputStream;
40+
import javax.annotation.Nullable;
41+
42+
/**
43+
* Stateful per-request implementation of {@link ResumableUploadFuture}. Manages payload chunking,
44+
* stream offsets, and session tracking.
45+
*
46+
* @param <RequestT> request type
47+
* @param <ResponseT> response type
48+
*/
49+
@BetaApi
50+
public class ResumableUploadFutureImpl<RequestT, ResponseT> extends AbstractApiFuture<ResponseT>
51+
implements ResumableUploadFuture<ResponseT> {
52+
53+
private final ResumableUploadClient resumableUploadClient;
54+
@Nullable private final RequestT request;
55+
@Nullable private final String initialSessionUrl;
56+
private final InputStream payload;
57+
private final ResumableUploadCallSettings settings;
58+
private final ApiCallContext context;
59+
60+
private volatile String uploadSessionUrl;
61+
private volatile long committedOffset = 0L;
62+
private volatile ApiFuture<?> inFlightFuture;
63+
64+
public ResumableUploadFutureImpl(
65+
ResumableUploadClient resumableUploadClient,
66+
RequestT request,
67+
InputStream payload,
68+
ResumableUploadCallSettings settings,
69+
ApiCallContext context) {
70+
this.resumableUploadClient = Preconditions.checkNotNull(resumableUploadClient);
71+
this.request = Preconditions.checkNotNull(request);
72+
this.initialSessionUrl = null;
73+
this.payload = Preconditions.checkNotNull(payload);
74+
this.settings = settings;
75+
this.context = context;
76+
}
77+
78+
public ResumableUploadFutureImpl(
79+
ResumableUploadClient resumableUploadClient,
80+
String sessionUrl,
81+
InputStream payload,
82+
ResumableUploadCallSettings settings,
83+
ApiCallContext context) {
84+
this.resumableUploadClient = Preconditions.checkNotNull(resumableUploadClient);
85+
this.request = null;
86+
this.initialSessionUrl = Preconditions.checkNotNull(sessionUrl);
87+
this.payload = Preconditions.checkNotNull(payload);
88+
this.settings = settings;
89+
this.context = context;
90+
}
91+
92+
public void start() {
93+
if (initialSessionUrl != null) {
94+
this.uploadSessionUrl = initialSessionUrl;
95+
uploadNextChunk();
96+
} else {
97+
initiateSessionAndUpload();
98+
}
99+
}
100+
101+
private void initiateSessionAndUpload() {
102+
ApiFuture<ResumableUploadSession> sessionFuture =
103+
resumableUploadClient.<RequestT>startUploadCallable().futureCall(request, context);
104+
this.inFlightFuture = sessionFuture;
105+
106+
ApiFutures.addCallback(
107+
sessionFuture,
108+
new ApiFutureCallback<ResumableUploadSession>() {
109+
@Override
110+
public void onSuccess(ResumableUploadSession session) {
111+
uploadSessionUrl = session.getUploadUrl();
112+
uploadNextChunk();
113+
}
114+
115+
@Override
116+
public void onFailure(Throwable t) {
117+
setException(t);
118+
}
119+
},
120+
MoreExecutors.directExecutor());
121+
}
122+
123+
private void uploadNextChunk() {
124+
try {
125+
int chunkSize = settings != null ? settings.getChunkSize() : 8 * 1024 * 1024;
126+
byte[] buffer = new byte[chunkSize];
127+
int bytesRead = payload.read(buffer);
128+
129+
if (bytesRead == -1) {
130+
bytesRead = 0;
131+
}
132+
133+
byte[] chunkData;
134+
if (bytesRead < chunkSize) {
135+
chunkData = new byte[bytesRead];
136+
System.arraycopy(buffer, 0, chunkData, 0, bytesRead);
137+
} else {
138+
chunkData = buffer;
139+
}
140+
141+
ChunkUploadRequest chunkRequest =
142+
new ChunkUploadRequest(uploadSessionUrl, chunkData, committedOffset, -1L, bytesRead == 0);
143+
144+
ApiFuture<ChunkUploadResponse> chunkFuture =
145+
resumableUploadClient.uploadChunkCallable().futureCall(chunkRequest, context);
146+
this.inFlightFuture = chunkFuture;
147+
148+
ApiFutures.addCallback(
149+
chunkFuture,
150+
new ApiFutureCallback<ChunkUploadResponse>() {
151+
@Override
152+
public void onSuccess(ChunkUploadResponse response) {
153+
committedOffset = response.getCommittedOffset();
154+
if (response.getResponseBody() != null) {
155+
@SuppressWarnings("unchecked")
156+
ResponseT result = (ResponseT) response.getResponseBody();
157+
set(result);
158+
} else {
159+
uploadNextChunk();
160+
}
161+
}
162+
163+
@Override
164+
public void onFailure(Throwable t) {
165+
setException(t);
166+
}
167+
},
168+
MoreExecutors.directExecutor());
169+
} catch (Throwable t) {
170+
setException(t);
171+
}
172+
}
173+
174+
@Override
175+
public String getUploadSessionUrl() {
176+
return uploadSessionUrl;
177+
}
178+
179+
public long getCommittedOffset() {
180+
return committedOffset;
181+
}
182+
183+
@Override
184+
public boolean cancel(boolean mayInterruptIfRunning) {
185+
if (inFlightFuture != null) {
186+
inFlightFuture.cancel(mayInterruptIfRunning);
187+
}
188+
return super.cancel(mayInterruptIfRunning);
189+
}
190+
}

0 commit comments

Comments
 (0)