@@ -78,6 +78,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
7878 private final long deadlineNanos ;
7979 private final ApiCallContext callContext ;
8080 private final ClientContext clientContext ;
81+ private final UploadProgressTracker progressTracker ;
8182 private final SettableApiFuture <ResponseT > result = SettableApiFuture .create ();
8283 private volatile @ Nullable ApiFuture <?> currentChunkFuture ;
8384
@@ -88,7 +89,8 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
8889 InputStream payload ,
8990 ResumableUploadCallSettings settings ,
9091 long deadlineNanos ,
91- ClientContext clientContext ) {
92+ ClientContext clientContext ,
93+ UploadProgressTracker progressTracker ) {
9294 this .uploadChunkCallable =
9395 checkNotNull (uploadChunkCallable , "uploadChunkCallable must not be null" );
9496 this .queryStatusCallable =
@@ -99,6 +101,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
99101 this .deadlineNanos = deadlineNanos ;
100102 this .clientContext = checkNotNull (clientContext , "clientContext must not be null" );
101103 this .callContext = clientContext .getDefaultCallContext ();
104+ this .progressTracker = checkNotNull (progressTracker , "progressTracker must not be null" );
102105 this .buffer = new RewindableStreamBuffer (payload , settings .getChunkSize (), uploadUrl );
103106 }
104107
@@ -148,7 +151,7 @@ public void onSuccess(ChunkUploadResponse<ResponseT> response) {
148151 return ;
149152 }
150153 if (response .getUploadStatus () == ResumableUploadStatus .UNKNOWN ) {
151- recover ();
154+ recover (null );
152155 } else {
153156 onChunkUploaded (response );
154157 }
@@ -162,7 +165,7 @@ public void onFailure(Throwable t) {
162165 Category category =
163166 ResumableUploadErrorClassifier .classify (t , ResumableUploadCommand .UPLOAD );
164167 if (category == Category .RECOVERABLE ) {
165- recover ();
168+ recover (t );
166169 } else {
167170 // Category.TRANSIENT errors reaching here have already exhausted their retry budget
168171 // in the underlying RetryingCallable and become fatal per protocol specification.
@@ -176,7 +179,14 @@ public void onFailure(Throwable t) {
176179 }
177180 }
178181
179- private void recover () {
182+ /**
183+ * Queries the session for the server's committed offset and resumes transmission from it.
184+ *
185+ * @param cause the error that triggered recovery, or null if the server acknowledged the chunk
186+ * without an upload status header
187+ */
188+ private void recover (@ Nullable Throwable cause ) {
189+ progressTracker .onRecovering (cause );
180190 try {
181191 // Dispatch the query status call and register the in-flight future for cancellation.
182192 ApiFuture <QueryStatusResponse <ResponseT >> queryFuture =
@@ -239,22 +249,29 @@ private void handleQueryResponse(QueryStatusResponse<ResponseT> queryResponse)
239249 "Incomplete query status response did not include a committed offset for upload URL: "
240250 + uploadUrl );
241251 }
252+ progressTracker .onOffsetReceived (committedOffset );
242253 buffer .realignTo (committedOffset );
243254 dispatchCurrentChunk ();
244255 }
245256
246257 private void onChunkUploaded (ChunkUploadResponse <ResponseT > response ) {
247- long nextOffset = buffer .getBufferBaseOffset () + buffer .getPayloadLength ();
248- if (response .getUploadStatus () == ResumableUploadStatus .FINAL ) {
249- result .set (response .getResponse ());
250- } else if (buffer .isFinal ()) {
251- result .setException (
252- new IllegalStateException (
253- "Upload stream ended and final chunk was transmitted, but server returned"
254- + " incomplete status for upload URL: "
255- + uploadUrl ));
256- } else {
257- chunkExecutor .execute (() -> transmitChunk (nextOffset ));
258+ try {
259+ long nextOffset = buffer .getBufferBaseOffset () + buffer .getPayloadLength ();
260+ if (response .getUploadStatus () == ResumableUploadStatus .FINAL ) {
261+ progressTracker .onChunkUploaded (nextOffset );
262+ result .set (response .getResponse ());
263+ } else if (buffer .isFinal ()) {
264+ result .setException (
265+ new IllegalStateException (
266+ "Upload stream ended and final chunk was transmitted, but server returned"
267+ + " incomplete status for upload URL: "
268+ + uploadUrl ));
269+ } else {
270+ progressTracker .onChunkUploaded (nextOffset );
271+ chunkExecutor .execute (() -> transmitChunk (nextOffset ));
272+ }
273+ } catch (Throwable t ) {
274+ result .setException (t );
258275 }
259276 }
260277
0 commit comments