Add DedicatedThreadAsyncMediaCodecAdapter

The DedicatedThreadAsyncMediaCodecAdapter is an
asynchronous MediaCodecAdapter that routes callback
to a separate Thread.

PiperOrigin-RevId: 285397368

library/core/src/main/java/com/google/android/exoplayer2/mediacodec/DedicatedThreadAsyncMediaCodecAdapter.java

@@ -0,0 +1,238 @@
+/*
+ * Copyright (C) 2019 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.android.exoplayer2.mediacodec;
+
+import android.media.MediaCodec;
+import android.media.MediaFormat;
+import android.os.Handler;
+import android.os.HandlerThread;
+import androidx.annotation.IntDef;
+import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+import androidx.annotation.RequiresApi;
+import androidx.annotation.VisibleForTesting;
+import com.google.android.exoplayer2.C;
+import com.google.android.exoplayer2.util.Assertions;
+import com.google.android.exoplayer2.util.Util;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+
+/**
+ * A {@link MediaCodecAdapter} that operates the underlying {@link MediaCodec} in asynchronous mode
+ * and routes {@link MediaCodec.Callback} callbacks on a dedicated Thread that is managed
+ * internally.
+ *
+ * <p>After creating an instance, you need to call {@link #start()} to start the internal Thread.
+ */
+@RequiresApi(23)
+/* package */ final class DedicatedThreadAsyncMediaCodecAdapter extends MediaCodec.Callback
+    implements MediaCodecAdapter {
+
+  @IntDef({State.CREATED, State.STARTED, State.SHUT_DOWN})
+  private @interface State {
+    int CREATED = 0;
+    int STARTED = 1;
+    int SHUT_DOWN = 2;
+  }
+
+  private final MediaCodecAsyncCallback mediaCodecAsyncCallback;
+  private final MediaCodec codec;
+  private final HandlerThread handlerThread;
+  @MonotonicNonNull private Handler handler;
+  private long pendingFlushCount;
+  private @State int state;
+  private Runnable onCodecStart;
+  @Nullable private IllegalStateException internalException;
+
+  /**
+   * Creates an instance that wraps the specified {@link MediaCodec}.
+   *
+   * @param codec The {@link MediaCodec} to wrap.
+   * @param trackType One of {@link C#TRACK_TYPE_AUDIO} or {@link C#TRACK_TYPE_VIDEO}. Used for
+   *     labelling the internal Thread accordingly.
+   * @throws IllegalArgumentException If {@code trackType} is not one of {@link C#TRACK_TYPE_AUDIO}
+   *     or {@link C#TRACK_TYPE_VIDEO}.
+   */
+  /* package */ DedicatedThreadAsyncMediaCodecAdapter(MediaCodec codec, int trackType) {
+    this(codec, new HandlerThread(createThreadLabel(trackType)));
+  }
+
+  @VisibleForTesting
+  /* package */ DedicatedThreadAsyncMediaCodecAdapter(
+      MediaCodec codec, HandlerThread handlerThread) {
+    mediaCodecAsyncCallback = new MediaCodecAsyncCallback();
+    this.codec = codec;
+    this.handlerThread = handlerThread;
+    state = State.CREATED;
+    onCodecStart = codec::start;
+  }
+
+  /**
+   * Starts the operation of the instance.
+   *
+   * <p>After a call to this method, make sure to call {@link #shutdown()} to terminate the internal
+   * Thread. You can only call this method once during the lifetime of this instance; calling this
+   * method again will throw an {@link IllegalStateException}.
+   *
+   * @throws IllegalStateException If this method has been called already.
+   */
+  public synchronized void start() {
+    Assertions.checkState(state == State.CREATED);
+
+    handlerThread.start();
+    handler = new Handler(handlerThread.getLooper());
+    codec.setCallback(this, handler);
+    state = State.STARTED;
+  }
+
+  @Override
+  public synchronized int dequeueInputBufferIndex() {
+    Assertions.checkState(state == State.STARTED);
+
+    if (isFlushing()) {
+      return MediaCodec.INFO_TRY_AGAIN_LATER;
+    } else {
+      maybeThrowException();
+      return mediaCodecAsyncCallback.dequeueInputBufferIndex();
+    }
+  }
+
+  @Override
+  public synchronized int dequeueOutputBufferIndex(MediaCodec.BufferInfo bufferInfo) {
+    Assertions.checkState(state == State.STARTED);
+
+    if (isFlushing()) {
+      return MediaCodec.INFO_TRY_AGAIN_LATER;
+    } else {
+      maybeThrowException();
+      return mediaCodecAsyncCallback.dequeueOutputBufferIndex(bufferInfo);
+    }
+  }
+
+  @Override
+  public synchronized MediaFormat getOutputFormat() {
+    Assertions.checkState(state == State.STARTED);
+
+    return mediaCodecAsyncCallback.getOutputFormat();
+  }
+
+  @Override
+  public synchronized void flush() {
+    Assertions.checkState(state == State.STARTED);
+
+    codec.flush();
+    ++pendingFlushCount;
+    Util.castNonNull(handler).post(this::onFlushCompleted);
+  }
+
+  @Override
+  public synchronized void shutdown() {
+    if (state == State.STARTED) {
+      handlerThread.quit();
+      mediaCodecAsyncCallback.flush();
+    }
+
+    state = State.SHUT_DOWN;
+  }
+
+  @Override
+  public synchronized void onInputBufferAvailable(@NonNull MediaCodec codec, int index) {
+    mediaCodecAsyncCallback.onInputBufferAvailable(codec, index);
+  }
+
+  @Override
+  public synchronized void onOutputBufferAvailable(
+      @NonNull MediaCodec codec, int index, @NonNull MediaCodec.BufferInfo info) {
+    mediaCodecAsyncCallback.onOutputBufferAvailable(codec, index, info);
+  }
+
+  @Override
+  public synchronized void onError(
+      @NonNull MediaCodec codec, @NonNull MediaCodec.CodecException e) {
+    mediaCodecAsyncCallback.onError(codec, e);
+  }
+
+  @Override
+  public synchronized void onOutputFormatChanged(
+      @NonNull MediaCodec codec, @NonNull MediaFormat format) {
+    mediaCodecAsyncCallback.onOutputFormatChanged(codec, format);
+  }
+
+  @VisibleForTesting
+  /* package */ void onMediaCodecError(IllegalStateException e) {
+    mediaCodecAsyncCallback.onMediaCodecError(e);
+  }
+
+  @VisibleForTesting
+  /* package */ void setOnCodecStart(Runnable onCodecStart) {
+    this.onCodecStart = onCodecStart;
+  }
+
+  private synchronized void onFlushCompleted() {
+    if (state != State.STARTED) {
+      // The adapter has been shutdown.
+      return;
+    }
+
+    --pendingFlushCount;
+    if (pendingFlushCount > 0) {
+      // Another flush() has been called.
+      return;
+    } else if (pendingFlushCount < 0) {
+      // This should never happen.
+      internalException = new IllegalStateException();
+      return;
+    }
+
+    mediaCodecAsyncCallback.flush();
+    try {
+      onCodecStart.run();
+    } catch (IllegalStateException e) {
+      internalException = e;
+    } catch (Exception e) {
+      internalException = new IllegalStateException(e);
+    }
+  }
+
+  private synchronized boolean isFlushing() {
+    return pendingFlushCount > 0;
+  }
+
+  private synchronized void maybeThrowException() {
+    maybeThrowInternalException();
+    mediaCodecAsyncCallback.maybeThrowMediaCodecException();
+  }
+
+  private synchronized void maybeThrowInternalException() {
+    if (internalException != null) {
+      IllegalStateException e = internalException;
+      internalException = null;
+      throw e;
+    }
+  }
+
+  private static String createThreadLabel(int trackType) {
+    StringBuilder labelBuilder = new StringBuilder("MediaCodecAsyncAdapter:");
+    if (trackType == C.TRACK_TYPE_AUDIO) {
+      labelBuilder.append("Audio");
+    } else if (trackType == C.TRACK_TYPE_VIDEO) {
+      labelBuilder.append("Video");
+    } else {
+      labelBuilder.append("Unknown(").append(trackType).append(")");
+    }
+    return labelBuilder.toString();
+  }
+}

library/core/src/main/java/com/google/android/exoplayer2/mediacodec/MediaCodecRenderer.java

@@ -195,7 +195,8 @@ public abstract class MediaCodecRenderer extends BaseRenderer {
   @Retention(RetentionPolicy.SOURCE)
   @IntDef({
     MediaCodecOperationMode.SYNCHRONOUS,
-    MediaCodecOperationMode.ASYNCHRONOUS_PLAYBACK_THREAD
+    MediaCodecOperationMode.ASYNCHRONOUS_PLAYBACK_THREAD,
+    MediaCodecOperationMode.ASYNCHRONOUS_DEDICATED_THREAD
   })
   public @interface MediaCodecOperationMode {
 
@@ -206,6 +207,11 @@ public abstract class MediaCodecRenderer extends BaseRenderer {
      * callbacks to the playback Thread.
      */
     int ASYNCHRONOUS_PLAYBACK_THREAD = 1;
+    /**
+     * Operates the {@link MediaCodec} in asynchronous mode and routes {@link MediaCodec.Callback}
+     * callbacks to a dedicated Thread.
+     */
+    int ASYNCHRONOUS_DEDICATED_THREAD = 2;
   }
 
   /** Indicates no codec operating rate should be set. */
@@ -472,6 +478,11 @@ public abstract class MediaCodecRenderer extends BaseRenderer {
    *           routed to the Playback Thread. This mode requires API level ≥ 21; if the API level
    *           is ≤ 20, the operation mode will be set to {@link
    *           MediaCodecOperationMode#SYNCHRONOUS}.
+   *       <li>{@link MediaCodecOperationMode#ASYNCHRONOUS_DEDICATED_THREAD}: The {@link MediaCodec}
+   *           will operate in asynchronous mode and {@link MediaCodec.Callback} callbacks will be
+   *           routed to a dedicated Thread. This mode requires API level &ge; 23; if the API level
+   *           is &le; 22, the operation mode will be set to {@link
+   *           MediaCodecOperationMode#SYNCHRONOUS}.
    *     </ul>
    *     By default, the operation mode is set to {@link MediaCodecOperationMode#SYNCHRONOUS}.
    */
@@ -943,6 +954,10 @@ public abstract class MediaCodecRenderer extends BaseRenderer {
       if (mediaCodecOperationMode == MediaCodecOperationMode.ASYNCHRONOUS_PLAYBACK_THREAD
           && Util.SDK_INT >= 21) {
         codecAdapter = new AsynchronousMediaCodecAdapter(codec);
+      } else if (mediaCodecOperationMode == MediaCodecOperationMode.ASYNCHRONOUS_DEDICATED_THREAD
+          && Util.SDK_INT >= 23) {
+        codecAdapter = new DedicatedThreadAsyncMediaCodecAdapter(codec, getTrackType());
+        ((DedicatedThreadAsyncMediaCodecAdapter) codecAdapter).start();
       } else {
         codecAdapter = new SynchronousMediaCodecAdapter(codec, getDequeueOutputBufferTimeoutUs());
       }

library/core/src/test/java/com/google/android/exoplayer2/mediacodec/DedicatedThreadAsyncMediaCodecAdapterTest.java

@@ -0,0 +1,437 @@
+/*
+ * Copyright (C) 2019 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.android.exoplayer2.mediacodec;
+
+import static com.google.android.exoplayer2.mediacodec.MediaCodecTestUtils.areEqual;
+import static com.google.android.exoplayer2.mediacodec.MediaCodecTestUtils.waitUntilAllEventsAreExecuted;
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
+import static org.robolectric.Shadows.shadowOf;
+
+import android.media.MediaCodec;
+import android.media.MediaFormat;
+import android.os.Handler;
+import android.os.HandlerThread;
+import android.os.Looper;
+import androidx.test.ext.junit.runners.AndroidJUnit4;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.robolectric.shadows.ShadowLooper;
+
+/** Unit tests for {@link DedicatedThreadAsyncMediaCodecAdapter}. */
+@RunWith(AndroidJUnit4.class)
+public class DedicatedThreadAsyncMediaCodecAdapterTest {
+  private DedicatedThreadAsyncMediaCodecAdapter adapter;
+  private MediaCodec codec;
+  private TestHandlerThread handlerThread;
+  private MediaCodec.BufferInfo bufferInfo = null;
+
+  @Before
+  public void setup() throws IOException {
+    codec = MediaCodec.createByCodecName("h264");
+    handlerThread = new TestHandlerThread("TestHandlerThread");
+    adapter = new DedicatedThreadAsyncMediaCodecAdapter(codec, handlerThread);
+    bufferInfo = new MediaCodec.BufferInfo();
+  }
+
+  @After
+  public void tearDown() {
+    adapter.shutdown();
+    assertThat(TestHandlerThread.INSTANCES_STARTED.get()).isEqualTo(0);
+  }
+
+  @Test
+  public void startAndShutdown_works() {
+    adapter.start();
+    adapter.shutdown();
+  }
+
+  @Test
+  public void start_calledTwice_throwsException() {
+    adapter.start();
+    try {
+      adapter.start();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_withoutStart_throwsException() {
+    try {
+      adapter.dequeueInputBufferIndex();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_afterShutdown_throwsException() {
+    adapter.start();
+    adapter.shutdown();
+    try {
+      adapter.dequeueInputBufferIndex();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_withAfterFlushFailed_throwsException()
+      throws InterruptedException {
+    adapter.setOnCodecStart(
+        () -> {
+          throw new IllegalStateException("codec#start() exception");
+        });
+    adapter.start();
+    adapter.flush();
+
+    assertThat(
+            waitUntilAllEventsAreExecuted(
+                handlerThread.getLooper(), /* time= */ 5, TimeUnit.SECONDS))
+        .isTrue();
+    try {
+      adapter.dequeueInputBufferIndex();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_withoutInputBuffer_returnsTryAgainLater() {
+    adapter.start();
+
+    assertThat(adapter.dequeueInputBufferIndex()).isEqualTo(MediaCodec.INFO_TRY_AGAIN_LATER);
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_withInputBuffer_returnsInputBuffer() {
+    adapter.start();
+    adapter.onInputBufferAvailable(codec, 0);
+
+    assertThat(adapter.dequeueInputBufferIndex()).isEqualTo(0);
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_withPendingFlush_returnsTryAgainLater() {
+    adapter.start();
+    adapter.onInputBufferAvailable(codec, 0);
+    adapter.flush();
+
+    assertThat(adapter.dequeueInputBufferIndex()).isEqualTo(MediaCodec.INFO_TRY_AGAIN_LATER);
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_withFlushCompletedAndInputBuffer_returnsInputBuffer()
+      throws InterruptedException {
+    // Disable calling codec.start() after flush to avoid receiving buffers from the
+    // shadow codec impl
+    adapter.setOnCodecStart(() -> {});
+    adapter.start();
+    Looper looper = handlerThread.getLooper();
+    Handler handler = new Handler(looper);
+    // Enqueue 10 callbacks from codec
+    for (int i = 0; i < 10; i++) {
+      int bufferIndex = i;
+      handler.post(() -> adapter.onInputBufferAvailable(codec, bufferIndex));
+    }
+    adapter.flush(); // Enqueues a flush event after the onInputBufferAvailable callbacks
+    // Enqueue another onInputBufferAvailable after the flush event
+    handler.post(() -> adapter.onInputBufferAvailable(codec, 10));
+
+    // Wait until all tasks have been handled
+    assertThat(waitUntilAllEventsAreExecuted(looper, /* time= */ 5, TimeUnit.SECONDS)).isTrue();
+    assertThat(adapter.dequeueInputBufferIndex()).isEqualTo(10);
+  }
+
+  @Test
+  public void dequeueInputBufferIndex_withMediaCodecError_throwsException() {
+    adapter.start();
+    adapter.onMediaCodecError(new IllegalStateException("error from codec"));
+
+    try {
+      adapter.dequeueInputBufferIndex();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_withoutStart_throwsException() {
+    try {
+      adapter.dequeueOutputBufferIndex(bufferInfo);
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_afterShutdown_throwsException() {
+    adapter.start();
+    adapter.shutdown();
+    try {
+      adapter.dequeueOutputBufferIndex(bufferInfo);
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_withInternalException_throwsException()
+      throws InterruptedException {
+    adapter.setOnCodecStart(
+        () -> {
+          throw new RuntimeException("codec#start() exception");
+        });
+    adapter.start();
+    adapter.flush();
+
+    assertThat(
+            waitUntilAllEventsAreExecuted(
+                handlerThread.getLooper(), /* time= */ 5, TimeUnit.SECONDS))
+        .isTrue();
+    try {
+      adapter.dequeueOutputBufferIndex(bufferInfo);
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_withoutInputBuffer_returnsTryAgainLater() {
+    adapter.start();
+
+    assertThat(adapter.dequeueOutputBufferIndex(bufferInfo))
+        .isEqualTo(MediaCodec.INFO_TRY_AGAIN_LATER);
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_withOutputBuffer_returnsOutputBuffer() {
+    adapter.start();
+    MediaCodec.BufferInfo enqueuedBufferInfo = new MediaCodec.BufferInfo();
+    adapter.onOutputBufferAvailable(codec, 0, enqueuedBufferInfo);
+
+    assertThat(adapter.dequeueOutputBufferIndex((bufferInfo))).isEqualTo(0);
+    assertThat(areEqual(bufferInfo, enqueuedBufferInfo)).isTrue();
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_withPendingFlush_returnsTryAgainLater() {
+    adapter.start();
+    adapter.dequeueOutputBufferIndex(bufferInfo);
+    adapter.flush();
+
+    assertThat(adapter.dequeueOutputBufferIndex(bufferInfo))
+        .isEqualTo(MediaCodec.INFO_TRY_AGAIN_LATER);
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_withFlushCompletedAndOutputBuffer_returnsOutputBuffer()
+      throws InterruptedException {
+    adapter.start();
+    Looper looper = handlerThread.getLooper();
+    Handler handler = new Handler(looper);
+    // Enqueue 10 callbacks from codec
+    for (int i = 0; i < 10; i++) {
+      int bufferIndex = i;
+      MediaCodec.BufferInfo outBufferInfo = new MediaCodec.BufferInfo();
+      outBufferInfo.presentationTimeUs = i;
+      handler.post(() -> adapter.onOutputBufferAvailable(codec, bufferIndex, outBufferInfo));
+    }
+    adapter.flush(); // Enqueues a flush event after the onOutputBufferAvailable callbacks
+    // Enqueue another onOutputBufferAvailable after the flush event
+    MediaCodec.BufferInfo lastBufferInfo = new MediaCodec.BufferInfo();
+    lastBufferInfo.presentationTimeUs = 10;
+    handler.post(() -> adapter.onOutputBufferAvailable(codec, 10, lastBufferInfo));
+
+    // Wait until all tasks have been handled
+    assertThat(waitUntilAllEventsAreExecuted(looper, /* time= */ 5, TimeUnit.SECONDS)).isTrue();
+    assertThat(adapter.dequeueOutputBufferIndex(bufferInfo)).isEqualTo(10);
+    assertThat(areEqual(bufferInfo, lastBufferInfo)).isTrue();
+  }
+
+  @Test
+  public void dequeueOutputBufferIndex_withMediaCodecError_throwsException() {
+    adapter.start();
+    adapter.onMediaCodecError(new IllegalStateException("error from codec"));
+
+    try {
+      adapter.dequeueOutputBufferIndex(bufferInfo);
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void getOutputFormat_withoutStart_throwsException() {
+    try {
+      adapter.getOutputFormat();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void getOutputFormat_afterShutdown_throwsException() {
+    adapter.start();
+    adapter.shutdown();
+    try {
+      adapter.getOutputFormat();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void getOutputFormat_withoutFormatReceived_throwsException() {
+    adapter.start();
+
+    try {
+      adapter.getOutputFormat();
+      fail();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void getOutputFormat_withMultipleFormats_returnsCorrectFormat() {
+    adapter.start();
+    MediaFormat[] formats = new MediaFormat[10];
+    for (int i = 0; i < formats.length; i++) {
+      formats[i] = new MediaFormat();
+      adapter.onOutputFormatChanged(codec, formats[i]);
+    }
+
+    for (int i = 0; i < 10; i++) {
+      assertThat(adapter.dequeueOutputBufferIndex(bufferInfo))
+          .isEqualTo(MediaCodec.INFO_OUTPUT_FORMAT_CHANGED);
+      assertThat(adapter.getOutputFormat()).isEqualTo(formats[i]);
+      // A subsequent call to getOutputFormat() should return the previously fetched format
+      assertThat(adapter.getOutputFormat()).isEqualTo(formats[i]);
+    }
+    assertThat(adapter.dequeueOutputBufferIndex(bufferInfo))
+        .isEqualTo(MediaCodec.INFO_TRY_AGAIN_LATER);
+  }
+
+  @Test
+  public void getOutputFormat_afterFlush_returnsPreviousFormat() throws InterruptedException {
+    MediaFormat format = new MediaFormat();
+    adapter.start();
+    adapter.onOutputFormatChanged(codec, format);
+
+    assertThat(adapter.dequeueOutputBufferIndex(bufferInfo))
+        .isEqualTo(MediaCodec.INFO_OUTPUT_FORMAT_CHANGED);
+    assertThat(adapter.getOutputFormat()).isEqualTo(format);
+
+    adapter.flush();
+    assertThat(
+            waitUntilAllEventsAreExecuted(
+                handlerThread.getLooper(), /* time= */ 5, TimeUnit.SECONDS))
+        .isTrue();
+    assertThat(adapter.getOutputFormat()).isEqualTo(format);
+  }
+
+  @Test
+  public void flush_withoutStarted_throwsException() {
+    try {
+      adapter.flush();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void flush_afterShutdown_throwsException() {
+    adapter.start();
+    adapter.shutdown();
+    try {
+      adapter.flush();
+    } catch (IllegalStateException expected) {
+    }
+  }
+
+  @Test
+  public void flush_multipleTimes_onlyLastFlushExecutes() throws InterruptedException {
+    AtomicInteger onCodecStartCount = new AtomicInteger(0);
+    adapter.setOnCodecStart(() -> onCodecStartCount.incrementAndGet());
+    adapter.start();
+    Looper looper = handlerThread.getLooper();
+    Handler handler = new Handler(looper);
+    handler.post(() -> adapter.onInputBufferAvailable(codec, 0));
+    adapter.flush(); // Enqueues a flush event
+    handler.post(() -> adapter.onInputBufferAvailable(codec, 2));
+    AtomicInteger milestoneCount = new AtomicInteger(0);
+    handler.post(() -> milestoneCount.incrementAndGet());
+    adapter.flush(); // Enqueues a second flush event
+    handler.post(() -> adapter.onInputBufferAvailable(codec, 3));
+
+    // Progress the looper until the milestoneCount is increased - first flush event
+    // should have been a no-op
+    ShadowLooper shadowLooper = shadowOf(looper);
+    while (milestoneCount.get() < 1) {
+      shadowLooper.runOneTask();
+    }
+    assertThat(onCodecStartCount.get()).isEqualTo(0);
+
+    assertThat(waitUntilAllEventsAreExecuted(looper, /* time= */ 5, TimeUnit.SECONDS)).isTrue();
+    assertThat(adapter.dequeueInputBufferIndex()).isEqualTo(3);
+    assertThat(onCodecStartCount.get()).isEqualTo(1);
+  }
+
+  @Test
+  public void flush_andImmediatelyShutdown_flushIsNoOp() throws InterruptedException {
+    AtomicInteger onCodecStartCount = new AtomicInteger(0);
+    adapter.setOnCodecStart(() -> onCodecStartCount.incrementAndGet());
+    adapter.start();
+    // Obtain looper when adapter is started
+    Looper looper = handlerThread.getLooper();
+    adapter.flush();
+    adapter.shutdown();
+
+    assertThat(waitUntilAllEventsAreExecuted(looper, 5, TimeUnit.SECONDS)).isTrue();
+    // only shutdown flushes the MediaCodecAsync handler
+    assertThat(onCodecStartCount.get()).isEqualTo(0);
+  }
+
+  private static class TestHandlerThread extends HandlerThread {
+    private static final AtomicLong INSTANCES_STARTED = new AtomicLong(0);
+
+    public TestHandlerThread(String name) {
+      super(name);
+    }
+
+    @Override
+    public synchronized void start() {
+      super.start();
+      INSTANCES_STARTED.incrementAndGet();
+    }
+
+    @Override
+    public boolean quit() {
+      boolean quit = super.quit();
+      if (quit) {
+        INSTANCES_STARTED.decrementAndGet();
+      }
+      return quit;
+    }
+  }
+}