Skip to content

Commit 6d2528d

Browse files
whoweswhowes
committed
feat(gax): implement MVP ResumableUploadCallable and ResumableUploadFuture
1 parent 0b2991e commit 6d2528d

6 files changed

Lines changed: 931 additions & 8 deletions

File tree

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

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -31,38 +31,49 @@
3131

3232
import com.google.api.core.BetaApi;
3333
import java.io.InputStream;
34+
import org.jspecify.annotations.NullMarked;
35+
import org.jspecify.annotations.Nullable;
3436

3537
/**
3638
* A ResumableUploadCallable is an API-transport-independent wrapper for the Resumable Upload
3739
* protocol. Operates directly on the request object and input stream payload.
3840
*
39-
* @param <RequestT> request type
40-
* @param <ResponseT> response type
41+
* @param <RequestT> the type of the initial request message that initiates the upload session
42+
* @param <ResponseT> the type of the final response message returned once the upload completes
4143
*/
4244
@BetaApi
45+
@NullMarked
4346
public abstract class ResumableUploadCallable<RequestT, ResponseT> {
4447

4548
protected ResumableUploadCallable() {}
4649

4750
/**
4851
* Performs a new resumable upload asynchronously.
4952
*
53+
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
54+
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
55+
* cancellation.
56+
*
5057
* @param request the request message
51-
* @param payload the data payload input stream
58+
* @param payload the data payload input stream to upload and close
5259
* @param settings call settings overrides; may be {@code null}
5360
* @return future for tracking and controlling the upload
5461
*/
5562
public abstract ResumableUploadFuture<ResponseT> futureCall(
56-
RequestT request, InputStream payload, ResumableUploadCallSettings settings);
63+
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings);
5764

5865
/**
5966
* Resumes an existing resumable upload session asynchronously using a saved session URL.
6067
*
68+
* <p>The provided {@code payload} stream is consumed asynchronously by the returned {@link
69+
* ResumableUploadFuture} and will be closed automatically upon completion, failure, or
70+
* cancellation.
71+
*
6172
* @param sessionUrl the upload session URL
62-
* @param payload the data payload input stream
73+
* @param payload the data payload input stream to upload and close
6374
* @param settings call settings overrides; may be {@code null}
6475
* @return future for tracking and controlling the upload
6576
*/
6677
public abstract ResumableUploadFuture<ResponseT> resumeCall(
67-
String sessionUrl, InputStream payload, ResumableUploadCallSettings settings);
78+
String sessionUrl, InputStream payload, @Nullable ResumableUploadCallSettings settings);
6879
}

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

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
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.core.ApiFuture;
35+
import com.google.api.core.ApiFutures;
36+
import com.google.api.core.BetaApi;
37+
import com.google.api.core.InternalApi;
38+
import com.google.api.gax.resumable.ResumableUploadClient;
39+
import com.google.api.gax.resumable.ResumableUploadSession;
40+
import java.io.InputStream;
41+
import java.util.concurrent.ScheduledExecutorService;
42+
import org.jspecify.annotations.NullMarked;
43+
import org.jspecify.annotations.Nullable;
44+
45+
/**
46+
* Concrete implementation of {@link ResumableUploadCallable} that delegates the end-to-end
47+
* management of a resumable upload session to {@link ResumableUploadFutureImpl}.
48+
*
49+
* @param <RequestT> the type of the initial request message that initiates the upload session
50+
* @param <ResponseT> the type of the final response message returned once the upload completes
51+
*/
52+
@BetaApi
53+
@InternalApi
54+
@NullMarked
55+
public class ResumableUploadCallableImpl<RequestT, ResponseT>
56+
extends ResumableUploadCallable<RequestT, ResponseT> {
57+
58+
private final ResumableUploadClient<RequestT, ResponseT> client;
59+
private final ResumableUploadCallSettings defaultCallSettings;
60+
private final ApiCallContext defaultCallContext;
61+
private final ScheduledExecutorService executor;
62+
63+
public ResumableUploadCallableImpl(
64+
ResumableUploadClient<RequestT, ResponseT> client,
65+
ResumableUploadCallSettings defaultCallSettings,
66+
ApiCallContext defaultCallContext,
67+
ScheduledExecutorService executor) {
68+
this.client = checkNotNull(client, "client must not be null");
69+
this.defaultCallSettings =
70+
checkNotNull(defaultCallSettings, "defaultCallSettings must not be null");
71+
this.defaultCallContext =
72+
checkNotNull(defaultCallContext, "defaultCallContext must not be null");
73+
this.executor = checkNotNull(executor, "executor must not be null");
74+
}
75+
76+
@Override
77+
public ResumableUploadFuture<ResponseT> futureCall(
78+
RequestT request, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
79+
checkNotNull(request, "request must not be null");
80+
checkNotNull(payload, "payload must not be null");
81+
ResumableUploadCallSettings effectiveSettings = defaultCallSettings.merge(settings);
82+
83+
ApiFuture<ResumableUploadSession> startFuture;
84+
try {
85+
startFuture = client.startUploadCallable().futureCall(request, defaultCallContext);
86+
} catch (Throwable t) {
87+
startFuture = ApiFutures.immediateFailedFuture(t);
88+
}
89+
90+
return ResumableUploadFutureImpl.create(
91+
startFuture,
92+
client.uploadChunkCallable(),
93+
payload,
94+
effectiveSettings,
95+
defaultCallContext,
96+
executor);
97+
}
98+
99+
@Override
100+
public ResumableUploadFuture<ResponseT> resumeCall(
101+
String sessionUrl, InputStream payload, @Nullable ResumableUploadCallSettings settings) {
102+
throw new UnsupportedOperationException("Session resumption is not yet implemented.");
103+
}
104+
}

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

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
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.core.ApiFuture;
35+
import com.google.api.core.ApiFutureCallback;
36+
import com.google.api.core.ApiFutures;
37+
import com.google.api.core.InternalApi;
38+
import com.google.api.gax.resumable.ChunkUploadRequest;
39+
import com.google.api.gax.resumable.ChunkUploadResponse;
40+
import com.google.common.io.ByteStreams;
41+
import java.io.IOException;
42+
import java.io.InputStream;
43+
import java.util.Arrays;
44+
import java.util.concurrent.CancellationException;
45+
import java.util.concurrent.ScheduledExecutorService;
46+
import org.jspecify.annotations.NullMarked;
47+
48+
/**
49+
* Coordinates chunk transmission steps of a resumable upload session.
50+
*
51+
* @param <ResponseT> the type of the final response message returned once the upload completes
52+
*/
53+
@InternalApi
54+
@NullMarked
55+
final class ResumableUploadChunkCoordinator<ResponseT> {
56+
57+
private static final byte[] EMPTY_PAYLOAD = new byte[0];
58+
59+
private final UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>>
60+
uploadChunkCallable;
61+
private final String uploadUrl;
62+
private final InputStream payload;
63+
private final int chunkSize;
64+
private final ApiCallContext callContext;
65+
private final ScheduledExecutorService executor;
66+
private final ResumableUploadFutureImpl<ResponseT> sessionFuture;
67+
68+
ResumableUploadChunkCoordinator(
69+
UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> uploadChunkCallable,
70+
String uploadUrl,
71+
InputStream payload,
72+
int chunkSize,
73+
ApiCallContext callContext,
74+
ScheduledExecutorService executor,
75+
ResumableUploadFutureImpl<ResponseT> sessionFuture) {
76+
this.uploadChunkCallable =
77+
checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
78+
this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
79+
this.payload = checkNotNull(payload, "payload must not be null");
80+
this.chunkSize = chunkSize;
81+
this.callContext = checkNotNull(callContext, "callContext must not be null");
82+
this.executor = checkNotNull(executor, "executor must not be null");
83+
this.sessionFuture = checkNotNull(sessionFuture, "sessionFuture must not be null");
84+
}
85+
86+
void start() {
87+
transmitChunk(0L);
88+
}
89+
90+
private void transmitChunk(long currentOffset) {
91+
// Abort if the session was already completed or canceled.
92+
if (sessionFuture.isDone()) {
93+
return;
94+
}
95+
96+
// Read the next chunk slice from the payload stream.
97+
byte[] buffer = new byte[chunkSize];
98+
int bytesRead;
99+
try {
100+
bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize);
101+
} catch (IOException e) {
102+
sessionFuture.fail(e);
103+
return;
104+
}
105+
106+
// Determine if this is the final chunk and build the chunk request.
107+
boolean isFinal = bytesRead < chunkSize;
108+
byte[] chunkPayload;
109+
if (bytesRead == chunkSize) {
110+
chunkPayload = buffer;
111+
} else if (bytesRead == 0) {
112+
chunkPayload = EMPTY_PAYLOAD;
113+
} else {
114+
chunkPayload = Arrays.copyOf(buffer, bytesRead);
115+
}
116+
117+
ChunkUploadRequest chunkRequest =
118+
ChunkUploadRequest.newBuilder()
119+
.setUploadUrl(uploadUrl)
120+
.setPayload(chunkPayload)
121+
.setOffset(currentOffset)
122+
.setFinal(isFinal)
123+
.build();
124+
125+
// Dispatch the chunk upload call and register the in-flight future for cancellation.
126+
long chunkLength = chunkPayload.length;
127+
try {
128+
ApiFuture<ChunkUploadResponse<ResponseT>> chunkFuture =
129+
uploadChunkCallable.futureCall(chunkRequest, callContext);
130+
sessionFuture.setInFlightFuture(chunkFuture);
131+
132+
// Asynchronously handle the response: complete, fail, or chain the next chunk.
133+
ApiFutures.addCallback(
134+
chunkFuture,
135+
new ApiFutureCallback<ChunkUploadResponse<ResponseT>>() {
136+
@Override
137+
public void onSuccess(ChunkUploadResponse<ResponseT> response) {
138+
if (sessionFuture.isDone()) {
139+
return;
140+
}
141+
long nextOffset = currentOffset + chunkLength;
142+
if (response.isComplete()) {
143+
sessionFuture.succeed(response.getResponse());
144+
} else if (isFinal) {
145+
sessionFuture.fail(
146+
new IllegalStateException(
147+
"Upload stream ended and final chunk was transmitted, but server returned"
148+
+ " incomplete status"));
149+
} else {
150+
transmitChunk(nextOffset);
151+
}
152+
}
153+
154+
@Override
155+
public void onFailure(Throwable t) {
156+
if (t instanceof CancellationException || sessionFuture.isDone()) {
157+
return;
158+
}
159+
sessionFuture.fail(t);
160+
}
161+
},
162+
executor);
163+
} catch (Throwable t) {
164+
sessionFuture.fail(t);
165+
}
166+
}
167+
}

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

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,15 +31,21 @@
3131

3232
import com.google.api.core.ApiFuture;
3333
import com.google.api.core.BetaApi;
34+
import org.jspecify.annotations.NullMarked;
35+
import org.jspecify.annotations.Nullable;
3436

3537
/**
3638
* A specialized {@link ApiFuture} for tracking and controlling an in-flight resumable upload.
3739
*
38-
* @param <ResponseT> response type
40+
* <p>The payload {@link java.io.InputStream} supplied when initiating the upload is managed by this
41+
* future and will be closed automatically upon completion, failure, or cancellation.
42+
*
43+
* @param <ResponseT> the type of the final response message returned once the upload completes
3944
*/
4045
@BetaApi
46+
@NullMarked
4147
public interface ResumableUploadFuture<ResponseT> extends ApiFuture<ResponseT> {
4248

4349
/** Returns the upload session URL, or {@code null} if session initiation is in progress. */
44-
String getUploadSessionUrl();
50+
@Nullable String getUploadSessionUrl();
4551
}

0 commit comments

Comments
 (0)