Push all Downloader networking onto the executor

Issue: Issue: #6978
PiperOrigin-RevId: 318710782

library/core/src/main/java/com/google/android/exoplayer2/offline/DefaultDownloaderFactory.java

@@ -18,6 +18,7 @@
 import android.net.Uri;
 import androidx.annotation.Nullable;
 import com.google.android.exoplayer2.upstream.cache.CacheDataSource;
+import com.google.android.exoplayer2.util.Assertions;
 import java.lang.reflect.Constructor;
 import java.util.List;
 import java.util.concurrent.Executor;
@@ -94,8 +95,8 @@ public DefaultDownloaderFactory(CacheDataSource.Factory cacheDataSourceFactory)
    */
   public DefaultDownloaderFactory(
       CacheDataSource.Factory cacheDataSourceFactory, Executor executor) {
-    this.cacheDataSourceFactory = cacheDataSourceFactory;
-    this.executor = executor;
+    this.cacheDataSourceFactory = Assertions.checkNotNull(cacheDataSourceFactory);
+    this.executor = Assertions.checkNotNull(executor);
   }
 
   @Override

library/core/src/main/java/com/google/android/exoplayer2/offline/DownloadManager.java

@@ -1331,7 +1331,7 @@ public void run() {
           }
         }
       } catch (InterruptedException e) {
-        // The task was canceled. Do nothing.
+        Thread.currentThread().interrupt();
       } catch (Exception e) {
         finalException = e;
       }

library/core/src/main/java/com/google/android/exoplayer2/offline/Downloader.java

@@ -18,6 +18,7 @@
 import androidx.annotation.Nullable;
 import com.google.android.exoplayer2.C;
 import java.io.IOException;
+import java.util.concurrent.CancellationException;
 
 /** Downloads and removes a piece of content. */
 public interface Downloader {
@@ -44,15 +45,29 @@ interface ProgressListener {
   /**
    * Downloads the content.
    *
+   * <p>If downloading fails, this method can be called again to resume the download. It cannot be
+   * called again after the download has been {@link #cancel canceled}.
+   *
+   * <p>If downloading is canceled whilst this method is executing, then it is expected that it will
+   * return reasonably quickly. However, there are no guarantees about how the method will return,
+   * meaning that it can return without throwing, or by throwing any of its documented exceptions.
+   * The caller must use its own knowledge about whether downloading has been canceled to determine
+   * whether this is why the method has returned, rather than relying on the method returning in a
+   * particular way.
+   *
    * @param progressListener A listener to receive progress updates, or {@code null}.
-   * @throws DownloadException Thrown if the content cannot be downloaded.
-   * @throws IOException If the download did not complete successfully.
+   * @throws IOException If the download failed to complete successfully.
+   * @throws InterruptedException If the download was interrupted.
+   * @throws CancellationException If the download was canceled.
    */
-  void download(@Nullable ProgressListener progressListener) throws IOException;
+  void download(@Nullable ProgressListener progressListener)
+      throws IOException, InterruptedException;
 
   /**
-   * Cancels the download operation and prevents future download operations from running. The caller
-   * should also interrupt the downloading thread immediately after calling this method.
+   * Permanently cancels the downloading by this downloader. The caller should also interrupt the
+   * downloading thread immediately after calling this method.
+   *
+   * <p>Once canceled, {@link #download} cannot be called again.
    */
   void cancel();
 

library/core/src/main/java/com/google/android/exoplayer2/offline/ProgressiveDownloader.java

@@ -25,16 +25,24 @@
 import com.google.android.exoplayer2.util.Assertions;
 import com.google.android.exoplayer2.util.PriorityTaskManager;
 import com.google.android.exoplayer2.util.PriorityTaskManager.PriorityTooLowException;
+import com.google.android.exoplayer2.util.RunnableFutureTask;
+import com.google.android.exoplayer2.util.Util;
 import java.io.IOException;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicBoolean;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 
 /** A downloader for progressive media streams. */
 public final class ProgressiveDownloader implements Downloader {
 
+  private final Executor executor;
   private final DataSpec dataSpec;
   private final CacheDataSource dataSource;
-  private final AtomicBoolean isCanceled;
+  @Nullable private final PriorityTaskManager priorityTaskManager;
+
+  @Nullable private ProgressListener progressListener;
+  private volatile @MonotonicNonNull RunnableFutureTask<Void, IOException> downloadRunnable;
+  private volatile boolean isCanceled;
 
   /** @deprecated Use {@link #ProgressiveDownloader(MediaItem, CacheDataSource.Factory)} instead. */
   @SuppressWarnings("deprecation")
@@ -84,6 +92,7 @@ public ProgressiveDownloader(
    */
   public ProgressiveDownloader(
       MediaItem mediaItem, CacheDataSource.Factory cacheDataSourceFactory, Executor executor) {
+    this.executor = Assertions.checkNotNull(executor);
     Assertions.checkNotNull(mediaItem.playbackProperties);
     dataSpec =
         new DataSpec.Builder()
@@ -92,40 +101,65 @@ public ProgressiveDownloader(
             .setFlags(DataSpec.FLAG_ALLOW_CACHE_FRAGMENTATION)
             .build();
     dataSource = cacheDataSourceFactory.createDataSourceForDownloading();
-    isCanceled = new AtomicBoolean();
+    priorityTaskManager = cacheDataSourceFactory.getUpstreamPriorityTaskManager();
   }
 
   @Override
-  public void download(@Nullable ProgressListener progressListener) throws IOException {
-    CacheWriter cacheWriter =
-        new CacheWriter(
-            dataSource,
-            dataSpec,
-            /* allowShortContent= */ false,
-            isCanceled,
-            /* temporaryBuffer= */ null,
-            progressListener == null ? null : new ProgressForwarder(progressListener));
+  public void download(@Nullable ProgressListener progressListener)
+      throws IOException, InterruptedException {
+    this.progressListener = progressListener;
+    if (downloadRunnable == null) {
+      CacheWriter cacheWriter =
+          new CacheWriter(
+              dataSource,
+              dataSpec,
+              /* allowShortContent= */ false,
+              /* temporaryBuffer= */ null,
+              this::onProgress);
+      downloadRunnable =
+          new RunnableFutureTask<Void, IOException>() {
+            @Override
+            protected Void doWork() throws IOException {
+              cacheWriter.cache();
+              return null;
+            }
+
+            @Override
+            protected void cancelWork() {
+              cacheWriter.cancel();
+            }
+          };
+    }
 
-    @Nullable PriorityTaskManager priorityTaskManager = dataSource.getUpstreamPriorityTaskManager();
     if (priorityTaskManager != null) {
       priorityTaskManager.add(C.PRIORITY_DOWNLOAD);
     }
     try {
       boolean finished = false;
-      while (!finished && !isCanceled.get()) {
+      while (!finished && !isCanceled) {
         if (priorityTaskManager != null) {
-          priorityTaskManager.proceed(dataSource.getUpstreamPriority());
+          priorityTaskManager.proceed(C.PRIORITY_DOWNLOAD);
         }
+        executor.execute(downloadRunnable);
         try {
-          cacheWriter.cache();
+          downloadRunnable.get();
           finished = true;
-        } catch (PriorityTooLowException e) {
-          // The next loop iteration will block until the task is able to proceed.
+        } catch (ExecutionException e) {
+          Throwable cause = Assertions.checkNotNull(e.getCause());
+          if (cause instanceof PriorityTooLowException) {
+            // The next loop iteration will block until the task is able to proceed.
+          } else if (cause instanceof IOException) {
+            throw (IOException) cause;
+          } else {
+            // The cause must be an uncaught Throwable type.
+            Util.sneakyThrow(cause);
+          }
         }
       }
-    } catch (InterruptedException e) {
-      // The download was canceled.
     } finally {
+      // If the main download thread was interrupted as part of cancelation, then it's possible that
+      // the runnable is still doing work. We need to wait until it's finished before returning.
+      downloadRunnable.blockUntilFinished();
       if (priorityTaskManager != null) {
         priorityTaskManager.remove(C.PRIORITY_DOWNLOAD);
       }
@@ -134,29 +168,26 @@ public void download(@Nullable ProgressListener progressListener) throws IOExcep
 
   @Override
   public void cancel() {
-    isCanceled.set(true);
+    isCanceled = true;
+    RunnableFutureTask<Void, IOException> downloadRunnable = this.downloadRunnable;
+    if (downloadRunnable != null) {
+      downloadRunnable.cancel(/* interruptIfRunning= */ true);
+    }
   }
 
   @Override
   public void remove() {
     dataSource.getCache().removeResource(dataSource.getCacheKeyFactory().buildCacheKey(dataSpec));
   }
 
-  private static final class ProgressForwarder implements CacheWriter.ProgressListener {
-
-    private final ProgressListener progressListener;
-
-    public ProgressForwarder(ProgressListener progressListener) {
-      this.progressListener = progressListener;
-    }
-
-    @Override
-    public void onProgress(long contentLength, long bytesCached, long newBytesCached) {
-      float percentDownloaded =
-          contentLength == C.LENGTH_UNSET || contentLength == 0
-              ? C.PERCENTAGE_UNSET
-              : ((bytesCached * 100f) / contentLength);
-      progressListener.onProgress(contentLength, bytesCached, percentDownloaded);
+  private void onProgress(long contentLength, long bytesCached, long newBytesCached) {
+    if (progressListener == null) {
+      return;
     }
+    float percentDownloaded =
+        contentLength == C.LENGTH_UNSET || contentLength == 0
+            ? C.PERCENTAGE_UNSET
+            : ((bytesCached * 100f) / contentLength);
+    progressListener.onProgress(contentLength, bytesCached, percentDownloaded);
   }
 }