Skip to content

Commit dac29ea

Browse files
committed
feat(gax): implement queryStatus in HttpJsonResumableUploadClient
1 parent b911782 commit dac29ea

5 files changed

Lines changed: 483 additions & 16 deletions

File tree

‎sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClient.java‎

Lines changed: 147 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@
3838
import com.google.api.core.SettableApiFuture;
3939
import com.google.api.gax.resumable.ChunkUploadRequest;
4040
import com.google.api.gax.resumable.ChunkUploadResponse;
41+
import com.google.api.gax.resumable.QueryStatusRequest;
42+
import com.google.api.gax.resumable.QueryStatusResponse;
4143
import com.google.api.gax.resumable.ResumableUploadClient;
4244
import com.google.api.gax.resumable.ResumableUploadSession;
4345
import com.google.api.gax.resumable.StartUploadRequest;
@@ -86,6 +88,9 @@ public final class HttpJsonResumableUploadClient implements ResumableUploadClien
8688

8789
private static final PathTemplate PATH_TEMPLATE = PathTemplate.create("{+path}");
8890

91+
private static final Map<String, List<String>> QUERY_STATUS_HEADERS =
92+
ImmutableMap.of(UPLOAD_COMMAND_HEADER, ImmutableList.of("query"));
93+
8994
private static final ApiMethodDescriptor<StartUploadRequest, String> START_UPLOAD_DESCRIPTOR =
9095
ApiMethodDescriptor.<StartUploadRequest, String>newBuilder()
9196
.setFullMethodName("ResumableUpload/StartUpload")
@@ -155,6 +160,41 @@ public PathTemplate getPathTemplate() {
155160
.setResponseParser(StringHttpResponseParser.create())
156161
.build();
157162

163+
private static final ApiMethodDescriptor<QueryStatusRequest, String> QUERY_STATUS_DESCRIPTOR =
164+
ApiMethodDescriptor.<QueryStatusRequest, String>newBuilder()
165+
.setFullMethodName("ResumableUpload/QueryStatus")
166+
.setHttpMethod(HttpMethods.POST)
167+
.setType(ApiMethodDescriptor.MethodType.UNARY)
168+
.setRequestFormatter(
169+
new HttpRequestFormatter<QueryStatusRequest>() {
170+
@Override
171+
public Map<String, List<String>> getQueryParamNames(QueryStatusRequest request) {
172+
return Collections.emptyMap();
173+
}
174+
175+
@Override
176+
public String getRequestBody(QueryStatusRequest request) {
177+
return "";
178+
}
179+
180+
@Override
181+
public HttpContent getHttpContent(QueryStatusRequest request) {
182+
return new EmptyContent();
183+
}
184+
185+
@Override
186+
public String getPath(QueryStatusRequest request) {
187+
return request.getUploadUrl();
188+
}
189+
190+
@Override
191+
public PathTemplate getPathTemplate() {
192+
return PATH_TEMPLATE;
193+
}
194+
})
195+
.setResponseParser(StringHttpResponseParser.create())
196+
.build();
197+
158198
private final ClientContext clientContext;
159199

160200
public static HttpJsonResumableUploadClient create(ClientContext clientContext) {
@@ -230,6 +270,32 @@ public ApiFuture<ChunkUploadResponse> futureCall(
230270
};
231271
}
232272

273+
@Override
274+
public UnaryCallable<QueryStatusRequest, QueryStatusResponse> queryStatusCallable() {
275+
return new UnaryCallable<QueryStatusRequest, QueryStatusResponse>() {
276+
@Override
277+
public ApiFuture<QueryStatusResponse> futureCall(
278+
QueryStatusRequest request, @Nullable ApiCallContext inputContext) {
279+
Preconditions.checkNotNull(request);
280+
HttpJsonCallContext context =
281+
(HttpJsonCallContext)
282+
HttpJsonCallContext.createDefault()
283+
.nullToSelf(clientContext.getDefaultCallContext())
284+
.merge(inputContext)
285+
.withExtraHeaders(QUERY_STATUS_HEADERS);
286+
287+
HttpJsonClientCall<QueryStatusRequest, String> clientCall =
288+
HttpJsonClientCalls.newCall(QUERY_STATUS_DESCRIPTOR, context);
289+
290+
SettableApiFuture<QueryStatusResponse> future = SettableApiFuture.create();
291+
HttpJsonClientCalls.startUnaryCall(
292+
clientCall, request, context, new QueryStatusResponseListener(future));
293+
294+
return future;
295+
}
296+
};
297+
}
298+
233299
private static class StartUploadResponseListener extends HttpJsonClientCall.Listener<String> {
234300

235301
private final SettableApiFuture<ResumableUploadSession> future;
@@ -308,22 +374,10 @@ private static class ChunkUploadResponseListener extends HttpJsonClientCall.List
308374

309375
@Override
310376
public void onHeaders(HttpJsonMetadata responseHeaders) {
311-
Map<String, Object> headers = responseHeaders.getHeaders();
312-
313-
String statusStr = HttpHeadersUtils.getFirstHeader(headers, UPLOAD_STATUS_HEADER);
314-
if (STATUS_FINAL.equalsIgnoreCase(statusStr)) {
315-
this.isComplete = true;
316-
}
317-
318-
String sizeReceivedStr =
319-
HttpHeadersUtils.getFirstHeader(headers, UPLOAD_SIZE_RECEIVED_HEADER);
320-
if (!Strings.isNullOrEmpty(sizeReceivedStr)) {
321-
try {
322-
this.committedOffset = Long.parseLong(sizeReceivedStr);
323-
} catch (NumberFormatException ignored) {
324-
// Ignore invalid/malformed size received header and fall back to local offset
325-
// calculation.
326-
}
377+
this.isComplete = isUploadFinal(responseHeaders);
378+
Long sizeReceived = parseSizeReceived(responseHeaders);
379+
if (sizeReceived != null) {
380+
this.committedOffset = sizeReceived;
327381
}
328382
}
329383

@@ -359,6 +413,83 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
359413
}
360414
}
361415

416+
private static class QueryStatusResponseListener extends HttpJsonClientCall.Listener<String> {
417+
418+
private final SettableApiFuture<QueryStatusResponse> future;
419+
private boolean isComplete = false;
420+
@Nullable private Long committedOffset = null;
421+
private String responseBody = "";
422+
423+
QueryStatusResponseListener(SettableApiFuture<QueryStatusResponse> future) {
424+
this.future = future;
425+
}
426+
427+
@Override
428+
public void onHeaders(HttpJsonMetadata responseHeaders) {
429+
this.isComplete = isUploadFinal(responseHeaders);
430+
this.committedOffset = parseSizeReceived(responseHeaders);
431+
}
432+
433+
@Override
434+
public void onMessage(@Nullable String message) {
435+
if (message != null) {
436+
this.responseBody = message;
437+
}
438+
}
439+
440+
@Override
441+
public void onClose(int statusCode, HttpJsonMetadata trailers) {
442+
try {
443+
if (statusCode >= 200 && statusCode < 300) {
444+
if (isComplete || committedOffset != null) {
445+
future.set(
446+
QueryStatusResponse.create(
447+
committedOffset != null ? committedOffset : 0L,
448+
isComplete,
449+
isComplete ? responseBody : ""));
450+
} else {
451+
future.setException(
452+
ApiExceptionFactory.createException(
453+
"Query status response did not contain valid X-Goog-Upload-Size-Received header",
454+
/* cause= */ null,
455+
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
456+
/* retryable= */ false));
457+
}
458+
} else {
459+
future.setException(
460+
createApiException(statusCode, trailers, "Failed to query upload status"));
461+
}
462+
} catch (Throwable t) {
463+
future.setException(
464+
ApiExceptionFactory.createException(
465+
"Internal error processing query status response",
466+
t,
467+
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
468+
/* retryable= */ false));
469+
}
470+
}
471+
}
472+
473+
private static boolean isUploadFinal(HttpJsonMetadata responseHeaders) {
474+
String statusStr =
475+
HttpHeadersUtils.getFirstHeader(responseHeaders.getHeaders(), UPLOAD_STATUS_HEADER);
476+
return STATUS_FINAL.equalsIgnoreCase(statusStr);
477+
}
478+
479+
@Nullable
480+
private static Long parseSizeReceived(HttpJsonMetadata responseHeaders) {
481+
String sizeReceivedStr =
482+
HttpHeadersUtils.getFirstHeader(responseHeaders.getHeaders(), UPLOAD_SIZE_RECEIVED_HEADER);
483+
if (!Strings.isNullOrEmpty(sizeReceivedStr)) {
484+
try {
485+
return Long.parseLong(sizeReceivedStr);
486+
} catch (NumberFormatException ignored) {
487+
// Unparseable header; return null and let the listener decide how to handle it.
488+
}
489+
}
490+
return null;
491+
}
492+
362493
private static ApiException createApiException(
363494
int statusCode, @Nullable HttpJsonMetadata trailers, String actionDescription) {
364495
Throwable cause = trailers != null ? trailers.getException() : null;

0 commit comments

Comments
 (0)