diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressListener.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressListener.java new file mode 100644 index 000000000000..ca79ec145f55 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressListener.java @@ -0,0 +1,50 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.rpc; + +import com.google.api.core.BetaApi; +import org.jspecify.annotations.NullMarked; + +/** A callback listener for observing progress and state transitions of a resumable upload. */ +@BetaApi +@FunctionalInterface +@NullMarked +public interface ResumableUploadProgressListener { + + /** + * Invoked when upload progress or state changes. + * + *

Cancellation via {@link ResumableUploadFuture#cancel(boolean)} can be invoked safely from + * within this callback. + * + * @param status the current status snapshot of the upload + */ + void onProgress(ResumableUploadStatus status); +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.java new file mode 100644 index 000000000000..c54f880b5b7d --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadStatus.java @@ -0,0 +1,101 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.rpc; + +import com.google.api.core.BetaApi; +import com.google.auto.value.AutoValue; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** Status snapshot of an ongoing or completed resumable upload session. */ +@BetaApi +@NullMarked +@AutoValue +public abstract class ResumableUploadStatus { + + /** The state of the resumable upload session. */ + public enum State { + /** Session initiation is in progress (acquiring upload session URL). */ + STARTING, + + /** The session initiation completed successfully. */ + STARTED, + + /** Transmitting chunk payloads to the server. */ + UPLOADING, + + /** A recoverable error occurred; querying server status and resynchronizing offset. */ + RECOVERING, + + /** The server query status succeeded and the committed offset was received. */ + OFFSET_RECEIVED, + + /** The upload was successfully finalized by the server. */ + FINALIZED, + + /** The upload failed unrecoverably or was cancelled. */ + FAILED + } + + /** + * Returns the negotiated upload session URI, or {@code null} if session initiation is pending. + */ + public abstract @Nullable String getUploadUrl(); + + /** Returns the number of bytes confirmed as uploaded to the server so far. */ + public abstract long getBytesUploaded(); + + /** Returns the current state of the upload session. */ + public abstract State getState(); + + /** Returns the exception that triggered recovery or caused failure, if any. */ + public abstract @Nullable Throwable getException(); + + public abstract Builder toBuilder(); + + public static Builder newBuilder() { + return new AutoValue_ResumableUploadStatus.Builder() + .setBytesUploaded(0L) + .setState(State.STARTING); + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setUploadUrl(@Nullable String uploadUrl); + + public abstract Builder setBytesUploaded(long bytesUploaded); + + public abstract Builder setState(State state); + + public abstract Builder setException(@Nullable Throwable exception); + + public abstract ResumableUploadStatus build(); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/UploadProgressTracker.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/UploadProgressTracker.java new file mode 100644 index 000000000000..a5a03dc82672 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/UploadProgressTracker.java @@ -0,0 +1,239 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.rpc; + +import static com.google.common.base.Preconditions.checkNotNull; + +import com.google.common.util.concurrent.MoreExecutors; +import com.google.errorprone.annotations.concurrent.GuardedBy; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executor; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +/** + * Thread-safe tracker and dispatcher for resumable upload progress and state transitions. + * + *

Enforces monotonic progress reporting, isolates listeners from upload pipeline failures, + * serializes callbacks per listener, and manages transition to terminal states. + */ +@NullMarked +class UploadProgressTracker { + + private static final class RegisteredListener { + final ResumableUploadProgressListener listener; + final Executor sequentialExecutor; + + RegisteredListener(ResumableUploadProgressListener listener, Executor executor) { + this.listener = listener; + this.sequentialExecutor = MoreExecutors.newSequentialExecutor(executor); + } + } + + private final Object lock = new Object(); + + @GuardedBy("lock") + private final List listeners = new ArrayList<>(); + + @GuardedBy("lock") + private ResumableUploadStatus currentStatus; + + @GuardedBy("lock") + private boolean terminal; + + @GuardedBy("lock") + private @Nullable String uploadSessionUrl; + + UploadProgressTracker() { + this.currentStatus = + ResumableUploadStatus.newBuilder() + .setState(ResumableUploadStatus.State.STARTING) + .setBytesUploaded(0L) + .build(); + } + + void addListener(ResumableUploadProgressListener listener, Executor executor) { + checkNotNull(listener, "listener must not be null"); + checkNotNull(executor, "executor must not be null"); + RegisteredListener entry = new RegisteredListener(listener, executor); + ResumableUploadStatus snapshot; + synchronized (lock) { + snapshot = this.currentStatus; + if (!terminal) { + listeners.add(entry); + } + } + entry.sequentialExecutor.execute(() -> dispatchSafely(listener, snapshot)); + } + + ResumableUploadStatus getStatus() { + synchronized (lock) { + return currentStatus; + } + } + + void onStarted(String uploadUrl) { + checkNotNull(uploadUrl, "uploadUrl must not be null"); + List snapshot; + ResumableUploadStatus status; + synchronized (lock) { + this.uploadSessionUrl = uploadUrl; + status = + currentStatus.toBuilder() + .setState(ResumableUploadStatus.State.STARTED) + .setUploadUrl(uploadUrl) + .build(); + snapshot = updateStatusLocked(status); + } + notifyListeners(snapshot, status); + } + + void onChunkUploaded(long bytesUploaded) { + List snapshot; + ResumableUploadStatus status; + synchronized (lock) { + if (terminal) { + return; + } + long bytes = Math.max(currentStatus.getBytesUploaded(), bytesUploaded); + status = + currentStatus.toBuilder() + .setState(ResumableUploadStatus.State.UPLOADING) + .setBytesUploaded(bytes) + .setUploadUrl(uploadSessionUrl) + .build(); + snapshot = updateStatusLocked(status); + } + notifyListeners(snapshot, status); + } + + void onRecovering(@Nullable Throwable cause) { + List snapshot; + ResumableUploadStatus status; + synchronized (lock) { + if (terminal) { + return; + } + status = + currentStatus.toBuilder() + .setState(ResumableUploadStatus.State.RECOVERING) + .setException(cause) + .setUploadUrl(uploadSessionUrl) + .build(); + snapshot = updateStatusLocked(status); + } + notifyListeners(snapshot, status); + } + + void onOffsetReceived(long committedOffset) { + List snapshot; + ResumableUploadStatus status; + synchronized (lock) { + if (terminal) { + return; + } + long bytes = Math.max(currentStatus.getBytesUploaded(), committedOffset); + status = + currentStatus.toBuilder() + .setState(ResumableUploadStatus.State.OFFSET_RECEIVED) + .setBytesUploaded(bytes) + .setUploadUrl(uploadSessionUrl) + .build(); + snapshot = updateStatusLocked(status); + } + notifyListeners(snapshot, status); + } + + void onFinalized(long totalBytes) { + List snapshot; + ResumableUploadStatus status; + synchronized (lock) { + if (terminal) { + return; + } + terminal = true; + long bytes = Math.max(currentStatus.getBytesUploaded(), totalBytes); + status = + currentStatus.toBuilder() + .setState(ResumableUploadStatus.State.FINALIZED) + .setBytesUploaded(bytes) + .setUploadUrl(uploadSessionUrl) + .build(); + snapshot = updateStatusLocked(status); + } + notifyListeners(snapshot, status); + } + + void onFailed(@Nullable Throwable error, @Nullable String sessionUrl) { + List snapshot; + ResumableUploadStatus status; + synchronized (lock) { + if (terminal) { + return; + } + terminal = true; + String url = sessionUrl != null ? sessionUrl : uploadSessionUrl; + status = + currentStatus.toBuilder() + .setState(ResumableUploadStatus.State.FAILED) + .setException(error) + .setUploadUrl(url) + .build(); + snapshot = updateStatusLocked(status); + } + notifyListeners(snapshot, status); + } + + @GuardedBy("lock") + private List updateStatusLocked(ResumableUploadStatus newStatus) { + this.currentStatus = newStatus; + if (newStatus.getUploadUrl() != null && this.uploadSessionUrl == null) { + this.uploadSessionUrl = newStatus.getUploadUrl(); + } + return new ArrayList<>(this.listeners); + } + + private void notifyListeners( + List targetListeners, ResumableUploadStatus status) { + for (RegisteredListener entry : targetListeners) { + entry.sequentialExecutor.execute(() -> dispatchSafely(entry.listener, status)); + } + } + + private static void dispatchSafely( + ResumableUploadProgressListener listener, ResumableUploadStatus status) { + try { + listener.onProgress(status); + } catch (Throwable ignored) { + // Listener exceptions are isolated from the upload pipeline + } + } +} diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadStatusTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadStatusTest.java new file mode 100644 index 000000000000..4a6507f1b2d6 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadStatusTest.java @@ -0,0 +1,81 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.rpc; + +import static com.google.common.truth.Truth.assertThat; + +import org.junit.jupiter.api.Test; + +class ResumableUploadStatusTest { + + @Test + void builder_defaults() { + ResumableUploadStatus status = ResumableUploadStatus.newBuilder().build(); + assertThat(status.getUploadUrl()).isNull(); + assertThat(status.getBytesUploaded()).isEqualTo(0L); + assertThat(status.getState()).isEqualTo(ResumableUploadStatus.State.STARTING); + assertThat(status.getException()).isNull(); + } + + @Test + void builder_explicitValuesAndToBuilder() { + Exception ex = new RuntimeException("test error"); + ResumableUploadStatus status = + ResumableUploadStatus.newBuilder() + .setUploadUrl("https://upload.url/session-1") + .setBytesUploaded(1024L) + .setState(ResumableUploadStatus.State.UPLOADING) + .setException(ex) + .build(); + + assertThat(status.getUploadUrl()).isEqualTo("https://upload.url/session-1"); + assertThat(status.getBytesUploaded()).isEqualTo(1024L); + assertThat(status.getState()).isEqualTo(ResumableUploadStatus.State.UPLOADING); + assertThat(status.getException()).isSameInstanceAs(ex); + + ResumableUploadStatus modified = + status.toBuilder() + .setState(ResumableUploadStatus.State.FINALIZED) + .setBytesUploaded(2048L) + .build(); + + assertThat(modified.getState()).isEqualTo(ResumableUploadStatus.State.FINALIZED); + assertThat(modified.getBytesUploaded()).isEqualTo(2048L); + assertThat(modified.getUploadUrl()).isEqualTo("https://upload.url/session-1"); + } + + @Test + void allStates_canBeRepresented() { + for (ResumableUploadStatus.State state : ResumableUploadStatus.State.values()) { + ResumableUploadStatus status = ResumableUploadStatus.newBuilder().setState(state).build(); + assertThat(status.getState()).isEqualTo(state); + } + } +} diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/UploadProgressTrackerTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/UploadProgressTrackerTest.java new file mode 100644 index 000000000000..385d2020a83b --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/UploadProgressTrackerTest.java @@ -0,0 +1,246 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.rpc; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.google.common.util.concurrent.MoreExecutors; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class UploadProgressTrackerTest { + + @Test + void testInitialSnapshotOnSubscribe() { + UploadProgressTracker tracker = new UploadProgressTracker(); + List statuses = new ArrayList<>(); + tracker.addListener(statuses::add, MoreExecutors.directExecutor()); + + assertThat(statuses).hasSize(1); + ResumableUploadStatus initial = statuses.get(0); + assertThat(initial.getState()).isEqualTo(ResumableUploadStatus.State.STARTING); + assertThat(initial.getBytesUploaded()).isEqualTo(0L); + assertThat(initial.getUploadUrl()).isNull(); + } + + @Test + void testMonotonicProgressEnforcement() { + UploadProgressTracker tracker = new UploadProgressTracker(); + tracker.onStarted("https://upload.url/test"); + + List byteUpdates = new ArrayList<>(); + tracker.addListener( + status -> byteUpdates.add(status.getBytesUploaded()), MoreExecutors.directExecutor()); + + // Advance to 100 bytes + tracker.onChunkUploaded(100L); + assertThat(byteUpdates).containsExactly(0L, 100L).inOrder(); + + // Out-of-order / smaller offset should not reduce reported bytes + tracker.onChunkUploaded(50L); + assertThat(byteUpdates).containsExactly(0L, 100L, 100L).inOrder(); + + // Larger offset advances + tracker.onChunkUploaded(200L); + assertThat(byteUpdates).containsExactly(0L, 100L, 100L, 200L).inOrder(); + } + + @Test + void testListenerExceptionSafety_doesNotDisruptSubsequentUpdates() { + UploadProgressTracker tracker = new UploadProgressTracker(); + tracker.onStarted("https://upload.url/test"); + + AtomicInteger errorCount = new AtomicInteger(0); + List safeReceived = new ArrayList<>(); + + // Listener 1 throws on every call + tracker.addListener( + status -> { + errorCount.incrementAndGet(); + throw new RuntimeException("boom from listener 1"); + }, + MoreExecutors.directExecutor()); + + // Listener 2 functions normally + tracker.addListener( + status -> safeReceived.add(status.getState()), MoreExecutors.directExecutor()); + + tracker.onChunkUploaded(50L); + tracker.onFinalized(50L); + + assertThat(errorCount.get()).isEqualTo(3); + assertThat(safeReceived) + .containsExactly( + ResumableUploadStatus.State.STARTED, + ResumableUploadStatus.State.UPLOADING, + ResumableUploadStatus.State.FINALIZED) + .inOrder(); + } + + @Test + void testTerminalState_noNotificationsAfterFinalized() { + UploadProgressTracker tracker = new UploadProgressTracker(); + tracker.onStarted("https://upload.url/test"); + + List received = new ArrayList<>(); + tracker.addListener(status -> received.add(status.getState()), MoreExecutors.directExecutor()); + + tracker.onChunkUploaded(100L); + tracker.onFinalized(100L); + + // Updates after finalized must be discarded + tracker.onChunkUploaded(150L); + tracker.onRecovering(new RuntimeException("should not appear")); + + assertThat(received) + .containsExactly( + ResumableUploadStatus.State.STARTED, + ResumableUploadStatus.State.UPLOADING, + ResumableUploadStatus.State.FINALIZED) + .inOrder(); + } + + @Test + void testTerminalState_noNotificationsAfterFailed() { + UploadProgressTracker tracker = new UploadProgressTracker(); + tracker.onStarted("https://upload.url/test"); + + List received = new ArrayList<>(); + tracker.addListener(status -> received.add(status.getState()), MoreExecutors.directExecutor()); + + tracker.onChunkUploaded(50L); + tracker.onFailed(new RuntimeException("failed"), "https://upload.url/test"); + + // Updates after failed must be discarded + tracker.onChunkUploaded(100L); + tracker.onFinalized(100L); + + assertThat(received) + .containsExactly( + ResumableUploadStatus.State.STARTED, + ResumableUploadStatus.State.UPLOADING, + ResumableUploadStatus.State.FAILED) + .inOrder(); + } + + @Test + void testCustomExecutorDispatch() throws Exception { + UploadProgressTracker tracker = new UploadProgressTracker(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + CountDownLatch latch = new CountDownLatch(2); + List states = new ArrayList<>(); + + tracker.addListener( + status -> { + synchronized (states) { + states.add(status.getState()); + } + latch.countDown(); + }, + executor); + + tracker.onStarted("https://upload.url/custom-executor"); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + + synchronized (states) { + assertThat(states) + .containsExactly( + ResumableUploadStatus.State.STARTING, ResumableUploadStatus.State.STARTED) + .inOrder(); + } + } finally { + executor.shutdownNow(); + } + } + + @Test + void testNoLockHeldDuringCallbackDispatch() throws Exception { + UploadProgressTracker tracker = new UploadProgressTracker(); + tracker.onStarted("https://upload.url/no-lock"); + + CountDownLatch listenerEntered = new CountDownLatch(1); + CountDownLatch probeCompleted = new CountDownLatch(1); + AtomicBoolean acquiredLockWhileInListener = new AtomicBoolean(false); + + tracker.addListener( + status -> { + if (status.getState() == ResumableUploadStatus.State.UPLOADING) { + listenerEntered.countDown(); + try { + if (probeCompleted.await(5, TimeUnit.SECONDS)) { + acquiredLockWhileInListener.set(true); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }, + MoreExecutors.directExecutor()); + + ExecutorService probeExecutor = Executors.newSingleThreadExecutor(); + try { + probeExecutor.submit( + () -> { + try { + if (listenerEntered.await(5, TimeUnit.SECONDS)) { + tracker.onChunkUploaded(200L); + probeCompleted.countDown(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + tracker.onChunkUploaded(100L); + + assertThat(probeCompleted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(acquiredLockWhileInListener.get()).isTrue(); + } finally { + probeExecutor.shutdownNow(); + } + } + + @Test + void testNullListenerThrowsNpe() { + UploadProgressTracker tracker = new UploadProgressTracker(); + assertThrows( + NullPointerException.class, + () -> tracker.addListener(null, MoreExecutors.directExecutor())); + } +}