Allow Injection of custom Executor in ProgressiveMediaSource

libraries/exoplayer/src/main/java/androidx/media3/exoplayer/source/ProgressiveMediaPeriod.java

@@ -70,6 +70,7 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.Executor;
 import org.checkerframework.checker.nullness.qual.EnsuresNonNull;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 
@@ -172,6 +173,7 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
    * @param continueLoadingCheckIntervalBytes The number of bytes that should be loaded between each
    *     invocation of {@link Callback#onContinueLoadingRequested(SequenceableLoader)}.
    * @param singleSampleDurationUs The duration of media with a single sample in microseconds.
+   * @param downloadExecutor An {@link Executor} for supplying the loader's thread.
    */
   // maybeFinishPrepare is not posted to the handler until initialization completes.
   @SuppressWarnings({"nullness:argument", "nullness:methodref.receiver.bound"})
@@ -187,7 +189,8 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
       Allocator allocator,
       @Nullable String customCacheKey,
       int continueLoadingCheckIntervalBytes,
-      long singleSampleDurationUs) {
+      long singleSampleDurationUs,
+      @Nullable Executor downloadExecutor) {
     this.uri = uri;
     this.dataSource = dataSource;
     this.drmSessionManager = drmSessionManager;
@@ -198,7 +201,8 @@ import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
     this.allocator = allocator;
     this.customCacheKey = customCacheKey;
     this.continueLoadingCheckIntervalBytes = continueLoadingCheckIntervalBytes;
-    loader = new Loader("ProgressiveMediaPeriod");
+    loader = downloadExecutor != null ?
+        new Loader(downloadExecutor) : new Loader("ProgressiveMediaPeriod");
     this.progressiveMediaExtractor = progressiveMediaExtractor;
     this.singleSampleDurationUs = singleSampleDurationUs;
     loadCondition = new ConditionVariable();

libraries/exoplayer/src/main/java/androidx/media3/exoplayer/source/ProgressiveMediaSource.java

@@ -37,7 +37,10 @@ import androidx.media3.exoplayer.upstream.LoadErrorHandlingPolicy;
 import androidx.media3.extractor.DefaultExtractorsFactory;
 import androidx.media3.extractor.Extractor;
 import androidx.media3.extractor.ExtractorsFactory;
+import com.google.common.base.Supplier;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import java.util.Optional;
+import java.util.concurrent.Executor;
 
 /**
  * Provides one period that loads data from a {@link Uri} and extracted using an {@link Extractor}.
@@ -64,6 +67,7 @@ public final class ProgressiveMediaSource extends BaseMediaSource
     private DrmSessionManagerProvider drmSessionManagerProvider;
     private LoadErrorHandlingPolicy loadErrorHandlingPolicy;
     private int continueLoadingCheckIntervalBytes;
+    @Nullable private Supplier<Executor> downloadExecutor;
 
     /**
      * Creates a new factory for {@link ProgressiveMediaSource}s.
@@ -154,6 +158,7 @@ public final class ProgressiveMediaSource extends BaseMediaSource
       this.drmSessionManagerProvider = drmSessionManagerProvider;
       this.loadErrorHandlingPolicy = loadErrorHandlingPolicy;
       this.continueLoadingCheckIntervalBytes = continueLoadingCheckIntervalBytes;
+      this.downloadExecutor = () -> null;
     }
 
     @CanIgnoreReturnValue
@@ -197,6 +202,20 @@ public final class ProgressiveMediaSource extends BaseMediaSource
       return this;
     }
 
+    /**
+     * Sets a supplier that can return an {@link Executor} that is used for loading the media. This
+     * is useful if the loading thread needs to be externally managed.
+     *
+     * @param downloadExecutor a {@link Supplier<Executor>} that provides an externally managed
+     *                         {@link Executor} for downloading and extraction.
+     * @return This factory, for convenience.
+     */
+    @CanIgnoreReturnValue
+    public Factory setDownloadExecutor(Supplier<Executor> downloadExecutor) {
+      this.downloadExecutor = downloadExecutor;
+      return this;
+    }
+
     /**
      * Returns a new {@link ProgressiveMediaSource} using the current parameters.
      *
@@ -213,7 +232,8 @@ public final class ProgressiveMediaSource extends BaseMediaSource
           progressiveMediaExtractorFactory,
           drmSessionManagerProvider.get(mediaItem),
           loadErrorHandlingPolicy,
-          continueLoadingCheckIntervalBytes);
+          continueLoadingCheckIntervalBytes,
+          downloadExecutor);
     }
 
     @Override
@@ -233,12 +253,12 @@ public final class ProgressiveMediaSource extends BaseMediaSource
   private final DrmSessionManager drmSessionManager;
   private final LoadErrorHandlingPolicy loadableLoadErrorHandlingPolicy;
   private final int continueLoadingCheckIntervalBytes;
+  @Nullable private final Supplier<Executor> downloadExecutor;
   private boolean timelineIsPlaceholder;
   private long timelineDurationUs;
   private boolean timelineIsSeekable;
   private boolean timelineIsLive;
   @Nullable private TransferListener transferListener;
-
   @GuardedBy("this")
   private MediaItem mediaItem;
 
@@ -248,7 +268,8 @@ public final class ProgressiveMediaSource extends BaseMediaSource
       ProgressiveMediaExtractor.Factory progressiveMediaExtractorFactory,
       DrmSessionManager drmSessionManager,
       LoadErrorHandlingPolicy loadableLoadErrorHandlingPolicy,
-      int continueLoadingCheckIntervalBytes) {
+      int continueLoadingCheckIntervalBytes,
+      @Nullable Supplier<Executor> downloadExecutor) {
     this.mediaItem = mediaItem;
     this.dataSourceFactory = dataSourceFactory;
     this.progressiveMediaExtractorFactory = progressiveMediaExtractorFactory;
@@ -257,6 +278,7 @@ public final class ProgressiveMediaSource extends BaseMediaSource
     this.continueLoadingCheckIntervalBytes = continueLoadingCheckIntervalBytes;
     this.timelineIsPlaceholder = true;
     this.timelineDurationUs = C.TIME_UNSET;
+    this.downloadExecutor = downloadExecutor;
   }
 
   @Override
@@ -312,7 +334,8 @@ public final class ProgressiveMediaSource extends BaseMediaSource
         allocator,
         localConfiguration.customCacheKey,
         continueLoadingCheckIntervalBytes,
-        Util.msToUs(localConfiguration.imageDurationMs));
+        Util.msToUs(localConfiguration.imageDurationMs),
+        downloadExecutor != null ? downloadExecutor.get() : null);
   }
 
   @Override

libraries/exoplayer/src/test/java/androidx/media3/exoplayer/source/ProgressiveMediaPeriodTest.java

@@ -19,6 +19,7 @@ import static androidx.media3.test.utils.robolectric.RobolectricUtil.runMainLoop
 import static com.google.common.truth.Truth.assertThat;
 
 import android.net.Uri;
+import androidx.annotation.Nullable;
 import androidx.media3.common.C;
 import androidx.media3.datasource.AssetDataSource;
 import androidx.media3.exoplayer.LoadingInfo;
@@ -34,6 +35,8 @@ import androidx.media3.extractor.mp4.Mp4Extractor;
 import androidx.media3.extractor.png.PngExtractor;
 import androidx.test.core.app.ApplicationProvider;
 import androidx.test.ext.junit.runners.AndroidJUnit4;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.Test;
@@ -67,8 +70,29 @@ public final class ProgressiveMediaPeriodTest {
     testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(extractor, C.TIME_UNSET);
   }
 
+  @Test
+  public void supplyingCustomDownloadExecutor_downloadsOnCustomThread() throws TimeoutException {
+    AtomicBoolean hasThreadRunBefore = new AtomicBoolean(false);
+    Executor executor =
+        Executors.newSingleThreadExecutor(
+            (r) -> new ExecutionTrackingThread(r, hasThreadRunBefore));
+
+    testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
+        new BundledExtractorsAdapter(Mp4Extractor.FACTORY), C.TIME_UNSET, executor);
+
+    assertThat(hasThreadRunBefore.get()).isTrue();
+  }
+
   private static void testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
       ProgressiveMediaExtractor extractor, long imageDurationUs) throws TimeoutException {
+    testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
+        extractor, imageDurationUs, null);
+  }
+
+  private static void testExtractorsUpdatesSourceInfoBeforeOnPreparedCallback(
+      ProgressiveMediaExtractor extractor,
+      long imageDurationUs,
+      @Nullable Executor executor) throws TimeoutException {
     AtomicBoolean sourceInfoRefreshCalled = new AtomicBoolean(false);
     ProgressiveMediaPeriod.Listener sourceInfoRefreshListener =
         (durationUs, isSeekable, isLive) -> sourceInfoRefreshCalled.set(true);
@@ -88,7 +112,8 @@ public final class ProgressiveMediaPeriodTest {
             new DefaultAllocator(/* trimOnReset= */ true, C.DEFAULT_BUFFER_SEGMENT_SIZE),
             /* customCacheKey= */ null,
             ProgressiveMediaSource.DEFAULT_LOADING_CHECK_INTERVAL_BYTES,
-            imageDurationUs);
+            imageDurationUs,
+            executor);
 
     AtomicBoolean prepareCallbackCalled = new AtomicBoolean(false);
     AtomicBoolean sourceInfoRefreshCalledBeforeOnPrepared = new AtomicBoolean(false);
@@ -111,4 +136,19 @@ public final class ProgressiveMediaPeriodTest {
 
     assertThat(sourceInfoRefreshCalledBeforeOnPrepared.get()).isTrue();
   }
+
+  private class ExecutionTrackingThread extends Thread {
+    private final AtomicBoolean hasRun;
+
+    public ExecutionTrackingThread(Runnable runnable, AtomicBoolean hasRun) {
+      super(runnable, "TestExecutionTrackingThread");
+      this.hasRun = hasRun;
+    }
+
+    @Override
+    public void run() {
+      hasRun.set(true);
+      super.run();
+    }
+  }
 }