Skip to content

Commit 1dc0716

Browse files
committed
feat(gax): implement startUpload in HttpJsonResumableUploadClient
1 parent d7e2be2 commit 1dc0716

2 files changed

Lines changed: 587 additions & 0 deletions

File tree

Lines changed: 217 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,217 @@
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.httpjson;
31+
32+
import com.google.api.client.http.HttpMethods;
33+
import com.google.api.core.ApiFuture;
34+
import com.google.api.core.InternalApi;
35+
import com.google.api.core.SettableApiFuture;
36+
import com.google.api.gax.resumable.ResumableUploadClient;
37+
import com.google.api.gax.resumable.ResumableUploadSession;
38+
import com.google.api.gax.rpc.ApiCallContext;
39+
import com.google.api.gax.rpc.ApiExceptionFactory;
40+
import com.google.api.gax.rpc.ClientContext;
41+
import com.google.api.gax.rpc.StatusCode;
42+
import com.google.api.gax.rpc.UnaryCallable;
43+
import com.google.common.base.Preconditions;
44+
import com.google.common.base.Strings;
45+
import com.google.common.collect.ImmutableList;
46+
import com.google.common.collect.ImmutableMap;
47+
import java.util.Collections;
48+
import java.util.List;
49+
import java.util.Map;
50+
import org.jspecify.annotations.NullMarked;
51+
import org.jspecify.annotations.Nullable;
52+
53+
/**
54+
* Implementation of {@link ResumableUploadClient} using HTTP/JSON transport.
55+
*
56+
* <p>Executes the low-level HTTP wire calls for managing resumable upload sessions.
57+
*
58+
* @param <RequestT> request type for starting an upload
59+
* @param <ResponseT> response type of the upload method
60+
*/
61+
@NullMarked
62+
@InternalApi
63+
public final class HttpJsonResumableUploadClient<RequestT, ResponseT>
64+
implements ResumableUploadClient<RequestT, ResponseT> {
65+
66+
private static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol";
67+
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
68+
private static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL";
69+
private static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity";
70+
71+
private static final Map<String, List<String>> START_UPLOAD_HEADERS =
72+
ImmutableMap.of(
73+
UPLOAD_PROTOCOL_HEADER, ImmutableList.of("resumable"),
74+
UPLOAD_COMMAND_HEADER, ImmutableList.of("start"));
75+
76+
private final ApiMethodDescriptor<RequestT, String> startUploadDescriptor;
77+
private final UnaryCallable<RequestT, ResumableUploadSession> startUploadCallable;
78+
79+
public static <RequestT, ResponseT> HttpJsonResumableUploadClient<RequestT, ResponseT> create(
80+
ClientContext clientContext, ApiMethodDescriptor<RequestT, ResponseT> methodDescriptor) {
81+
return new HttpJsonResumableUploadClient<>(clientContext, methodDescriptor);
82+
}
83+
84+
private HttpJsonResumableUploadClient(
85+
ClientContext clientContext, ApiMethodDescriptor<RequestT, ResponseT> methodDescriptor) {
86+
Preconditions.checkNotNull(clientContext);
87+
Preconditions.checkNotNull(methodDescriptor);
88+
89+
this.startUploadDescriptor =
90+
ApiMethodDescriptor.<RequestT, String>newBuilder()
91+
.setFullMethodName(methodDescriptor.getFullMethodName())
92+
.setHttpMethod(HttpMethods.POST)
93+
.setType(ApiMethodDescriptor.MethodType.UNARY)
94+
.setRequestFormatter(methodDescriptor.getRequestFormatter())
95+
.setResponseParser(ResumableUploadResponseParser.create())
96+
.build();
97+
98+
UnaryCallable<RequestT, ResumableUploadSession> rawStartUploadCallable =
99+
new UnaryCallable<RequestT, ResumableUploadSession>() {
100+
@Override
101+
public ApiFuture<ResumableUploadSession> futureCall(
102+
RequestT request, @Nullable ApiCallContext inputContext) {
103+
Preconditions.checkNotNull(request);
104+
HttpJsonCallContext context =
105+
(HttpJsonCallContext)
106+
HttpJsonCallContext.createDefault()
107+
.nullToSelf(clientContext.getDefaultCallContext())
108+
.merge(inputContext)
109+
.withExtraHeaders(START_UPLOAD_HEADERS);
110+
111+
HttpJsonClientCall<RequestT, String> clientCall =
112+
HttpJsonClientCalls.newCall(startUploadDescriptor, context);
113+
114+
SettableApiFuture<ResumableUploadSession> future = SettableApiFuture.create();
115+
HttpJsonClientCalls.startUnaryCall(
116+
clientCall, request, context, new StartUploadResponseListener(future));
117+
118+
return future;
119+
}
120+
};
121+
this.startUploadCallable =
122+
new HttpJsonExceptionCallable<>(rawStartUploadCallable, Collections.emptySet());
123+
}
124+
125+
@Override
126+
public UnaryCallable<RequestT, ResumableUploadSession> startUploadCallable() {
127+
return startUploadCallable;
128+
}
129+
130+
private static class StartUploadResponseListener extends HttpJsonClientCall.Listener<String> {
131+
132+
private final SettableApiFuture<ResumableUploadSession> future;
133+
@Nullable private String uploadUrl;
134+
private long chunkGranularity = 1L;
135+
@Nullable private Throwable headerParsingException;
136+
137+
StartUploadResponseListener(SettableApiFuture<ResumableUploadSession> future) {
138+
this.future = future;
139+
}
140+
141+
@Override
142+
public void onHeaders(HttpJsonMetadata responseHeaders) {
143+
Map<String, Object> headers = responseHeaders.getHeaders();
144+
145+
String url = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_URL_HEADER);
146+
if (!Strings.isNullOrEmpty(url)) {
147+
this.uploadUrl = url;
148+
}
149+
150+
String granularityStr = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_GRANULARITY_HEADER);
151+
if (!Strings.isNullOrEmpty(granularityStr)) {
152+
try {
153+
long parsed = Long.parseLong(granularityStr);
154+
if (parsed <= 0) {
155+
this.headerParsingException =
156+
ApiExceptionFactory.createException(
157+
"Start upload response contained non-positive chunk granularity header: "
158+
+ granularityStr,
159+
/* cause= */ null,
160+
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
161+
/* retryable= */ false);
162+
} else {
163+
this.chunkGranularity = parsed;
164+
}
165+
} catch (NumberFormatException e) {
166+
this.headerParsingException =
167+
ApiExceptionFactory.createException(
168+
"Start upload response contained invalid chunk granularity header: "
169+
+ granularityStr,
170+
e,
171+
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
172+
/* retryable= */ false);
173+
}
174+
}
175+
}
176+
177+
@Override
178+
public void onMessage(@Nullable String message) {
179+
// Response body is not needed for startUpload; session URL is in headers.
180+
}
181+
182+
@Override
183+
public void onClose(int statusCode, HttpJsonMetadata trailers) {
184+
try {
185+
if (statusCode >= 200 && statusCode < 300) {
186+
if (headerParsingException != null) {
187+
future.setException(headerParsingException);
188+
return;
189+
}
190+
if (!Strings.isNullOrEmpty(uploadUrl)) {
191+
future.set(
192+
ResumableUploadSession.newBuilder()
193+
.setUploadUrl(uploadUrl)
194+
.setChunkGranularity(chunkGranularity)
195+
.build());
196+
} else {
197+
future.setException(
198+
ApiExceptionFactory.createException(
199+
"Start upload response did not contain upload session URL header",
200+
/* cause= */ null,
201+
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
202+
/* retryable= */ false));
203+
}
204+
} else {
205+
Throwable cause = trailers.getException();
206+
future.setException(
207+
cause != null
208+
? cause
209+
: new HttpJsonStatusRuntimeException(
210+
statusCode, "Failed to start upload with status code: " + statusCode, null));
211+
}
212+
} catch (Throwable t) {
213+
future.setException(t);
214+
}
215+
}
216+
}
217+
}

0 commit comments

Comments
 (0)