Add Kotlin extension function on Player

Given that `Player` interface is written in Java and is has a callback-based Listener interface, we need an adapter for the Kotlin-native world.

This change introduces a suspending function `listen` that creates a coroutine, in order to capture `Player.Events`.

PiperOrigin-RevId: 658478608

demos/compose/build.gradle

@@ -58,6 +58,12 @@ android {
     composeOptions {
         kotlinCompilerExtensionVersion = "1.5.3"
     }
+
+    testOptions {
+        unitTests {
+            includeAndroidResources = true
+        }
+    }
 }
 
 dependencies {
@@ -73,4 +79,9 @@ dependencies {
 
     // For detecting and debugging leaks only. LeakCanary is not needed for demo app to work.
     debugImplementation 'com.squareup.leakcanary:leakcanary-android:' + leakCanaryVersion
+
+    testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:' + kotlinxCoroutinesVersion
+    testImplementation 'org.robolectric:robolectric:' + robolectricVersion
+    testImplementation project(modulePrefix + 'test-utils')
+
 }

demos/compose/src/main/java/androidx/media3/demo/compose/PlayerExtensions.kt

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2024 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
+ *
+ *      https://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 androidx.media3.demo.compose
+
+import android.os.Looper
+import androidx.core.os.HandlerCompat
+import androidx.media3.common.Player
+import androidx.media3.common.Player.Events
+import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.coroutines.resumeWithException
+import kotlinx.coroutines.CancellableContinuation
+import kotlinx.coroutines.android.asCoroutineDispatcher
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlinx.coroutines.withContext
+
+/**
+ * Continuously listens to the [Player.Listener.onEvents] callback, passing the received
+ * [Player.Events] to the provided [onEvents] function.
+ *
+ * This function can be called from any thread. The [onEvents] function will be invoked on the
+ * thread associated with [Player.getApplicationLooper].
+ *
+ * If, during the execution of [onEvents], an exception is thrown, the coroutine corresponding to
+ * listening to the Player will be terminated. Any used resources will be cleaned up (e.g. removing
+ * of the listeners) and exception will be re-thrown right after the last suspension point.
+ *
+ * @param onEvents The function to handle player events.
+ * @return Nothing This function never returns normally. It will either continue indefinitely or
+ *   terminate due to an exception or cancellation.
+ */
+suspend fun Player.listen(onEvents: Player.(Events) -> Unit): Nothing {
+  if (Looper.myLooper() == applicationLooper) {
+    listenImpl(onEvents)
+  } else
+    withContext(HandlerCompat.createAsync(applicationLooper).asCoroutineDispatcher()) {
+      listenImpl(onEvents)
+    }
+}
+
+/**
+ * Implements the core listening logic for [Player.Events].
+ *
+ * This function creates a cancellable coroutine that listens to [Player.Events] in an infinite
+ * loop. The coroutine can be cancelled externally, or it can terminate if the [onEvents] lambda
+ * throws an exception.
+ *
+ * Given that `invokeOnCancellation` block can be called at any time, we provide the thread-safety
+ * guarantee by:
+ * * unregistering the callback (i.e. removing the listener) in the finally block to keep on the
+ *   calling context which was previously ensured to be the application thread
+ * * marking the listener as cancelled using `AtomicBoolean` to ensure that this value will be
+ *   visible immediately on any non-calling thread due to a memory barrier.
+ *
+ * A note on [callbackFlow] vs [suspendCancellableCoroutine]:
+ *
+ * Despite [callbackFlow] being recommended for a multi-shot API (like [Player]'s), a
+ * [suspendCancellableCoroutine] is a lower-level construct that allows us to overcome the
+ * limitations of [Flow]'s buffered dispatch. In our case, we will not be waiting for a particular
+ * callback to resume the continuation (i.e. the common single-shot use of
+ * [suspendCancellableCoroutine]), but rather handle incoming Events indefinitely. This approach
+ * controls the timing of dispatching events to the caller more tightly than [Flow]s. Such timing
+ * guarantees are critical for responding to events with frame-perfect timing and become more
+ * relevant in the context of front-end UI development (e.g. using Compose).
+ */
+private suspend fun Player.listenImpl(onEvents: Player.(Events) -> Unit): Nothing {
+  lateinit var listener: PlayerListener
+  try {
+    suspendCancellableCoroutine<Nothing> { continuation ->
+      listener = PlayerListener(onEvents, continuation)
+      continuation.invokeOnCancellation { listener.isCancelled.set(true) }
+      addListener(listener)
+    }
+  } finally {
+    removeListener(listener)
+  }
+}
+
+private class PlayerListener(
+  private val onEvents: Player.(Events) -> Unit,
+  private val continuation: CancellableContinuation<Nothing>,
+) : Player.Listener {
+
+  val isCancelled: AtomicBoolean = AtomicBoolean(false)
+
+  override fun onEvents(player: Player, events: Events) {
+    try {
+      if (!isCancelled.get()) {
+        player.onEvents(events)
+      }
+    } catch (t: Throwable) {
+      isCancelled.set(true)
+      continuation.resumeWithException(t)
+    }
+  }
+}

demos/compose/src/test/AndroidManifest.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!-- Copyright 2024 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
+
+          https://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.
+-->
+
+<manifest package="androidx.media3.demo.compose.test">
+  <uses-sdk/>
+</manifest>

demos/compose/src/test/java/androidx/media3/demo/compose/PlayerExtensionsTest.kt

@@ -0,0 +1,189 @@
+/*
+ * Copyright 2024 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
+ *
+ *      https://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 androidx.media3.demo.compose
+
+import android.content.Context
+import android.os.Handler
+import android.os.HandlerThread
+import android.os.Looper
+import androidx.media3.common.ForwardingPlayer
+import androidx.media3.common.Player
+import androidx.media3.exoplayer.ExoPlayer
+import androidx.media3.test.utils.TestExoPlayerBuilder
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.google.common.truth.Truth.assertThat
+import kotlin.coroutines.cancellation.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.android.asCoroutineDispatcher
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.Shadows.shadowOf
+
+/** Unit tests for Kotlin extension functions on the [Player]. */
+@RunWith(AndroidJUnit4::class)
+class PlayerExtensionsTest {
+
+  private val context = ApplicationProvider.getApplicationContext<Context>()
+
+  @Test
+  fun playerListen_receivesVolumeEvent() = runTest {
+    var volumeFromInsideOnEvents: Float? = null
+    val player: ExoPlayer = TestExoPlayerBuilder(context).build()
+    val listenJob = launch {
+      player.listen { events ->
+        if (Player.EVENT_VOLUME_CHANGED in events) {
+          volumeFromInsideOnEvents = player.volume
+        }
+      }
+    }
+    // Wait for the Player.Listener to be registered inside player.listen
+    testScheduler.runCurrent()
+
+    // Set the volume to a non-default value to trigger an event
+    player.volume = 0.5f
+
+    // Let the volume change propagate
+    shadowOf(Looper.getMainLooper()).idle()
+
+    assertThat(volumeFromInsideOnEvents).isEqualTo(0.5f)
+    listenJob.cancelAndJoin()
+  }
+
+  @Test
+  fun playerListen_withInternalCancel_cancelsCoroutineAndUnregistersListener() = runTest {
+    val player = PlayerWithListeners(TestExoPlayerBuilder(context).build())
+    val listenJob = launch {
+      player.listen { events ->
+        if (Player.EVENT_VOLUME_CHANGED in events) {
+          throw CancellationException()
+        }
+      }
+    }
+    // Wait for the Player.Listener to be registered inside player.listen
+    testScheduler.runCurrent()
+
+    assertThat(player.listeners.size).isEqualTo(1)
+
+    // Set the volume to a non-default value to trigger an event
+    player.volume = 0.5f
+    // Let the volume change propagate
+    shadowOf(Looper.getMainLooper()).idle()
+    // Let the CancellationException propagate and trigger listener removal
+    testScheduler.runCurrent()
+
+    assertThat(player.listeners.size).isEqualTo(0)
+    assertThat(listenJob.isCancelled).isTrue()
+    assertThat(listenJob.isCompleted).isTrue()
+  }
+
+  @Test
+  fun playerListen_withExternalCancel_unregistersListener() = runTest {
+    val player = PlayerWithListeners(TestExoPlayerBuilder(context).build())
+    val listenJob = launch { player.listen { _ -> } }
+    // Wait for the Player.Listener to be registered inside player.listen
+    testScheduler.runCurrent()
+
+    assertThat(player.listeners.size).isEqualTo(1)
+
+    listenJob.cancelAndJoin()
+
+    assertThat(player.listeners.size).isEqualTo(0)
+    assertThat(listenJob.isCancelled).isTrue()
+    assertThat(listenJob.isCompleted).isTrue()
+  }
+
+  @Test
+  fun playerListen_onEventsThrowsException_bubblesOutAndUnregistersListener() = runTest {
+    val player = PlayerWithListeners(TestExoPlayerBuilder(context).build())
+    val exceptionFromListen = async {
+      try {
+        player.listen { events ->
+          if (Player.EVENT_VOLUME_CHANGED in events) {
+            throw IllegalStateException("Volume event!")
+          }
+        }
+      } catch (expected: IllegalStateException) {
+        expected
+      }
+    }
+    // Wait for the Player.Listener to be registered inside player.listen
+    testScheduler.runCurrent()
+
+    assertThat(player.listeners.size).isEqualTo(1)
+
+    // Set the volume to a non-default value to trigger an event
+    player.volume = 0.5f
+    // Let the volume change propagate
+    shadowOf(Looper.getMainLooper()).idle()
+
+    assertThat(exceptionFromListen.await()).hasMessageThat().isEqualTo("Volume event!")
+    assertThat(player.listeners.size).isEqualTo(0)
+  }
+
+  @Test
+  fun playerListen_calledFromDifferentThread_receivesVolumeEvent() = runTest {
+    val applicationThread = HandlerThread("app-thread")
+    applicationThread.start()
+    val applicationHandler = Handler(applicationThread.looper)
+    // Construct the player on application thread != test thread
+    // This is where Player.Events will be delivered to
+    val player =
+      withContext(applicationHandler.asCoroutineDispatcher()) {
+        TestExoPlayerBuilder(context).build()
+      }
+    val volumeFromInsideOnEventsJob = CompletableDeferred<Float>()
+    val listenJob = launch {
+      // Start listening from test thread != application thread
+      // Player is accessed from a different thread to where it was created
+      player.listen { events ->
+        if (Player.EVENT_VOLUME_CHANGED in events) {
+          // Complete a Job of getting the new volume out of a forever listening loop with success
+          volumeFromInsideOnEventsJob.complete(player.volume)
+        }
+      }
+    }
+    // Wait for the Player.Listener to be registered inside player.listen
+    testScheduler.runCurrent()
+
+    // Set the volume to a non-default value to trigger an event
+    // Use the application thread where the Player was constructed
+    launch(applicationHandler.asCoroutineDispatcher()) { player.volume = 0.5f }
+
+    assertThat(volumeFromInsideOnEventsJob.await()).isEqualTo(0.5f)
+    listenJob.cancelAndJoin()
+    applicationThread.quit()
+  }
+
+  private class PlayerWithListeners(player: Player) : ForwardingPlayer(player) {
+    val listeners: MutableSet<Player.Listener> = HashSet()
+
+    override fun addListener(listener: Player.Listener) {
+      super.addListener(listener)
+      listeners.add(listener)
+    }
+
+    override fun removeListener(listener: Player.Listener) {
+      super.removeListener(listener)
+      listeners.remove(listener)
+    }
+  }
+}