@@ -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
@@ -87,7 +88,8 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
8788 String uploadUrl ,
8889 ResumableUploadCallSettings settings ,
8990 long deadlineNanos ,
90- ClientContext clientContext ) {
91+ ClientContext clientContext ,
92+ UploadProgressTracker progressTracker ) {
9193 this .uploadChunkCallable =
9294 checkNotNull (uploadChunkCallable , "uploadChunkCallable must not be null" );
9395 this .queryStatusCallable =
@@ -98,6 +100,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
98100 this .deadlineNanos = deadlineNanos ;
99101 this .clientContext = checkNotNull (clientContext , "clientContext must not be null" );
100102 this .callContext = clientContext .getDefaultCallContext ();
103+ this .progressTracker = checkNotNull (progressTracker , "progressTracker must not be null" );
101104 this .buffer = new RewindableStreamBuffer (payload , settings .getChunkSize (), uploadUrl );
102105 }
103106
@@ -168,7 +171,7 @@ private ApiFuture<ChunkUploadResponse<ResponseT>> executeChunkWithRecovery() {
168171 Category category =
169172 ResumableUploadErrorClassifier .classify (t , ResumableUploadCommand .UPLOAD );
170173 if (category == Category .RECOVERABLE ) {
171- return recover ();
174+ return recover (t );
172175 }
173176 // Category.TRANSIENT errors reaching here have already exhausted their retry budget
174177 // in the underlying RetryingCallable and become fatal per protocol specification.
@@ -179,14 +182,21 @@ private ApiFuture<ChunkUploadResponse<ResponseT>> executeChunkWithRecovery() {
179182 transmitted ,
180183 response -> {
181184 if (response .getUploadStatus () == ResumableUploadStatus .UNKNOWN ) {
182- return recover ();
185+ return recover (null );
183186 }
184187 return ApiFutures .immediateFuture (response );
185188 },
186189 chunkExecutor );
187190 }
188191
189- private ApiFuture <ChunkUploadResponse <ResponseT >> recover () {
192+ /**
193+ * Queries the session for the server's committed offset and resumes transmission from it.
194+ *
195+ * @param cause the error that triggered recovery, or null if the server acknowledged the chunk
196+ * without an upload status header
197+ */
198+ private ApiFuture <ChunkUploadResponse <ResponseT >> recover (@ Nullable Throwable cause ) {
199+ progressTracker .onRecovering (cause );
190200 return ApiFutures .transformAsync (
191201 queryStatusCallable .futureCall (QueryStatusRequest .create (uploadUrl ), queryCallContext ()),
192202 this ::resumeFrom ,
@@ -210,24 +220,31 @@ private ApiFuture<ChunkUploadResponse<ResponseT>> resumeFrom(
210220 + uploadUrl );
211221 }
212222 buffer .realignTo (committedOffset );
223+ progressTracker .onOffsetReceived (committedOffset );
213224 return executeChunkWithRecovery ();
214225 }
215226
216227 private void onChunkUploaded (ChunkUploadResponse <ResponseT > response ) {
217228 if (result .isDone ()) {
218229 return ;
219230 }
220- long nextOffset = buffer .getBufferBaseOffset () + buffer .getPayloadLength ();
221- if (response .getUploadStatus () == ResumableUploadStatus .FINAL ) {
222- result .set (response .getResponse ());
223- } else if (buffer .isFinal ()) {
224- result .setException (
225- new IllegalStateException (
226- "Upload stream ended and final chunk was transmitted, but server returned"
227- + " incomplete status for upload URL: "
228- + uploadUrl ));
229- } else {
230- chunkExecutor .execute (() -> transmitChunk (nextOffset ));
231+ try {
232+ long nextOffset = buffer .getBufferBaseOffset () + buffer .getPayloadLength ();
233+ if (response .getUploadStatus () == ResumableUploadStatus .FINAL ) {
234+ progressTracker .onChunkUploaded (nextOffset );
235+ result .set (response .getResponse ());
236+ } else if (buffer .isFinal ()) {
237+ result .setException (
238+ new IllegalStateException (
239+ "Upload stream ended and final chunk was transmitted, but server returned"
240+ + " incomplete status for upload URL: "
241+ + uploadUrl ));
242+ } else {
243+ progressTracker .onChunkUploaded (nextOffset );
244+ chunkExecutor .execute (() -> transmitChunk (nextOffset ));
245+ }
246+ } catch (Throwable t ) {
247+ result .setException (t );
231248 }
232249 }
233250
0 commit comments