Move listener handling to common util class.

ExoPlayerImpl and CastPlayer repeat the same logic. Moving the listener
and event handling to a common util class allows to reuse the same code
and add unit tests for this logic.

The change is a functional no-op.

PiperOrigin-RevId: 337812358

extensions/cast/src/main/java/com/google/android/exoplayer2/ext/cast/CastPlayer.java

@@ -35,6 +35,7 @@ import com.google.android.exoplayer2.trackselection.TrackSelection;
 import com.google.android.exoplayer2.trackselection.TrackSelectionArray;
 import com.google.android.exoplayer2.trackselection.TrackSelector;
 import com.google.android.exoplayer2.util.Assertions;
+import com.google.android.exoplayer2.util.ListenerSet;
 import com.google.android.exoplayer2.util.Log;
 import com.google.android.exoplayer2.util.MimeTypes;
 import com.google.android.gms.cast.CastStatusCodes;
@@ -50,12 +51,8 @@ import com.google.android.gms.cast.framework.media.RemoteMediaClient;
 import com.google.android.gms.cast.framework.media.RemoteMediaClient.MediaChannelResult;
 import com.google.android.gms.common.api.PendingResult;
 import com.google.android.gms.common.api.ResultCallback;
-import java.util.ArrayDeque;
-import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Iterator;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 import org.checkerframework.checker.nullness.qual.RequiresNonNull;
 
 /**
@@ -99,9 +96,7 @@ public final class CastPlayer extends BasePlayer {
   private final SeekResultCallback seekResultCallback;
 
   // Listeners and notification.
-  private final CopyOnWriteArrayList<ListenerHolder> listeners;
+  private final ListenerSet<Player.EventListener> listeners;
-  private final ArrayList<ListenerNotificationTask> notificationsBatch;
-  private final ArrayDeque<ListenerNotificationTask> ongoingNotificationsTasks;
   @Nullable private SessionAvailabilityListener sessionAvailabilityListener;
 
   // Internal state.
@@ -140,9 +135,7 @@ public final class CastPlayer extends BasePlayer {
     period = new Timeline.Period();
     statusListener = new StatusListener();
     seekResultCallback = new SeekResultCallback();
-    listeners = new CopyOnWriteArrayList<>();
+    listeners = new ListenerSet<>();
-    notificationsBatch = new ArrayList<>();
-    ongoingNotificationsTasks = new ArrayDeque<>();
 
     playWhenReady = new StateHolder<>(false);
     repeatMode = new StateHolder<>(REPEAT_MODE_OFF);
@@ -295,18 +288,12 @@ public final class CastPlayer extends BasePlayer {
 
   @Override
   public void addListener(EventListener listener) {
-    Assertions.checkNotNull(listener);
+    listeners.add(listener);
-    listeners.addIfAbsent(new ListenerHolder(listener));
   }
 
   @Override
   public void removeListener(EventListener listener) {
-    for (ListenerHolder listenerHolder : listeners) {
+    listeners.remove(listener);
-      if (listenerHolder.listener.equals(listener)) {
-        listenerHolder.release();
-        listeners.remove(listenerHolder);
-      }
-    }
   }
 
   @Override
@@ -418,7 +405,7 @@ public final class CastPlayer extends BasePlayer {
     // the local state will be updated to reflect the state reported by the Cast SDK.
     setPlayerStateAndNotifyIfChanged(
         playWhenReady, PLAY_WHEN_READY_CHANGE_REASON_USER_REQUEST, playbackState);
-    flushNotifications();
+    listeners.flushEvents();
     PendingResult<MediaChannelResult> pendingResult =
         playWhenReady ? remoteMediaClient.play() : remoteMediaClient.pause();
     this.playWhenReady.pendingResultCallback =
@@ -427,7 +414,7 @@ public final class CastPlayer extends BasePlayer {
           public void onResult(MediaChannelResult mediaChannelResult) {
             if (remoteMediaClient != null) {
               updatePlayerStateAndNotifyIfChanged(this);
-              flushNotifications();
+              listeners.flushEvents();
             }
           }
         };
@@ -458,13 +445,11 @@ public final class CastPlayer extends BasePlayer {
       pendingSeekCount++;
       pendingSeekWindowIndex = windowIndex;
       pendingSeekPositionMs = positionMs;
-      notificationsBatch.add(
+      listeners.queueEvent(listener -> listener.onPositionDiscontinuity(DISCONTINUITY_REASON_SEEK));
-          new ListenerNotificationTask(
-              listener -> listener.onPositionDiscontinuity(DISCONTINUITY_REASON_SEEK)));
     } else if (pendingSeekCount == 0) {
-      notificationsBatch.add(new ListenerNotificationTask(EventListener::onSeekProcessed));
+      listeners.queueEvent(EventListener::onSeekProcessed);
     }
-    flushNotifications();
+    listeners.flushEvents();
   }
 
   @Override
@@ -528,7 +513,7 @@ public final class CastPlayer extends BasePlayer {
     // operation to be perceived as synchronous by the user. When the operation reports a result,
     // the local state will be updated to reflect the state reported by the Cast SDK.
     setRepeatModeAndNotifyIfChanged(repeatMode);
-    flushNotifications();
+    listeners.flushEvents();
     PendingResult<MediaChannelResult> pendingResult =
         remoteMediaClient.queueSetRepeatMode(getCastRepeatMode(repeatMode), /* jsonObject= */ null);
     this.repeatMode.pendingResultCallback =
@@ -537,7 +522,7 @@ public final class CastPlayer extends BasePlayer {
           public void onResult(MediaChannelResult mediaChannelResult) {
             if (remoteMediaClient != null) {
               updateRepeatModeAndNotifyIfChanged(this);
-              flushNotifications();
+              listeners.flushEvents();
             }
           }
         };
@@ -662,8 +647,7 @@ public final class CastPlayer extends BasePlayer {
     updatePlayerStateAndNotifyIfChanged(/* resultCallback= */ null);
     boolean isPlaying = playbackState == Player.STATE_READY && playWhenReady.value;
     if (wasPlaying != isPlaying) {
-      notificationsBatch.add(
+      listeners.queueEvent(listener -> listener.onIsPlayingChanged(isPlaying));
-          new ListenerNotificationTask(listener -> listener.onIsPlayingChanged(isPlaying)));
     }
     updateRepeatModeAndNotifyIfChanged(/* resultCallback= */ null);
     updateTimelineAndNotifyIfChanged();
@@ -679,17 +663,14 @@ public final class CastPlayer extends BasePlayer {
     }
     if (this.currentWindowIndex != currentWindowIndex && pendingSeekCount == 0) {
       this.currentWindowIndex = currentWindowIndex;
-      notificationsBatch.add(
+      listeners.queueEvent(
-          new ListenerNotificationTask(
+          listener -> listener.onPositionDiscontinuity(DISCONTINUITY_REASON_PERIOD_TRANSITION));
-              listener ->
-                  listener.onPositionDiscontinuity(DISCONTINUITY_REASON_PERIOD_TRANSITION)));
     }
     if (updateTracksAndSelectionsAndNotifyIfChanged()) {
-      notificationsBatch.add(
+      listeners.queueEvent(
-          new ListenerNotificationTask(
+          listener -> listener.onTracksChanged(currentTrackGroups, currentTrackSelection));
-              listener -> listener.onTracksChanged(currentTrackGroups, currentTrackSelection)));
     }
-    flushNotifications();
+    listeners.flushEvents();
   }
 
   /**
@@ -728,11 +709,10 @@ public final class CastPlayer extends BasePlayer {
     if (updateTimeline()) {
       // TODO: Differentiate TIMELINE_CHANGE_REASON_PLAYLIST_CHANGED and
       //     TIMELINE_CHANGE_REASON_SOURCE_UPDATE [see internal: b/65152553].
-      notificationsBatch.add(
+      listeners.queueEvent(
-          new ListenerNotificationTask(
           listener ->
               listener.onTimelineChanged(
-                      currentTimeline, Player.TIMELINE_CHANGE_REASON_SOURCE_UPDATE)));
+                  currentTimeline, Player.TIMELINE_CHANGE_REASON_SOURCE_UPDATE));
     }
   }
 
@@ -851,8 +831,7 @@ public final class CastPlayer extends BasePlayer {
   private void setRepeatModeAndNotifyIfChanged(@Player.RepeatMode int repeatMode) {
     if (this.repeatMode.value != repeatMode) {
       this.repeatMode.value = repeatMode;
-      notificationsBatch.add(
+      listeners.queueEvent(listener -> listener.onRepeatModeChanged(repeatMode));
-          new ListenerNotificationTask(listener -> listener.onRepeatModeChanged(repeatMode)));
     }
   }
 
@@ -866,8 +845,7 @@ public final class CastPlayer extends BasePlayer {
     if (playWhenReadyChanged || playbackStateChanged) {
       this.playbackState = playbackState;
       this.playWhenReady.value = playWhenReady;
-      notificationsBatch.add(
+      listeners.queueEvent(
-          new ListenerNotificationTask(
           listener -> {
             listener.onPlayerStateChanged(playWhenReady, playbackState);
             if (playbackStateChanged) {
@@ -876,7 +854,7 @@ public final class CastPlayer extends BasePlayer {
             if (playWhenReadyChanged) {
               listener.onPlayWhenReadyChanged(playWhenReady, playWhenReadyChangeReason);
             }
-              }));
+          });
     }
   }
 
@@ -984,20 +962,6 @@ public final class CastPlayer extends BasePlayer {
     }
   }
 
-  private void flushNotifications() {
-    boolean recursiveNotification = !ongoingNotificationsTasks.isEmpty();
-    ongoingNotificationsTasks.addAll(notificationsBatch);
-    notificationsBatch.clear();
-    if (recursiveNotification) {
-      // This will be handled once the current notification task is finished.
-      return;
-    }
-    while (!ongoingNotificationsTasks.isEmpty()) {
-      ongoingNotificationsTasks.peekFirst().execute();
-      ongoingNotificationsTasks.removeFirst();
-    }
-  }
-
   private MediaQueueItem[] toMediaQueueItems(List<MediaItem> mediaItems) {
     MediaQueueItem[] mediaQueueItems = new MediaQueueItem[mediaItems.size()];
     for (int i = 0; i < mediaItems.size(); i++) {
@@ -1108,8 +1072,7 @@ public final class CastPlayer extends BasePlayer {
       if (--pendingSeekCount == 0) {
         pendingSeekWindowIndex = C.INDEX_UNSET;
         pendingSeekPositionMs = C.TIME_UNSET;
-        notificationsBatch.add(new ListenerNotificationTask(EventListener::onSeekProcessed));
+        listeners.sendEvent(EventListener::onSeekProcessed);
-        flushNotifications();
       }
     }
   }
@@ -1149,21 +1112,4 @@ public final class CastPlayer extends BasePlayer {
       return pendingResultCallback == resultCallback;
     }
   }
-
-  private final class ListenerNotificationTask {
-
-    private final Iterator<ListenerHolder> listenersSnapshot;
-    private final ListenerInvocation listenerInvocation;
-
-    private ListenerNotificationTask(ListenerInvocation listenerInvocation) {
-      this.listenersSnapshot = listeners.iterator();
-      this.listenerInvocation = listenerInvocation;
-    }
-
-    public void execute() {
-      while (listenersSnapshot.hasNext()) {
-        listenersSnapshot.next().invoke(listenerInvocation);
-      }
-    }
-  }
 }

library/core/src/main/java/com/google/android/exoplayer2/BasePlayer.java

@@ -249,58 +249,4 @@ public abstract class BasePlayer implements Player {
     @RepeatMode int repeatMode = getRepeatMode();
     return repeatMode == REPEAT_MODE_ONE ? REPEAT_MODE_OFF : repeatMode;
   }
-
-  /** Holds a listener reference. */
-  protected static final class ListenerHolder {
-
-    /**
-     * The listener on which {link #invoke} will execute {@link ListenerInvocation listener
-     * invocations}.
-     */
-    public final Player.EventListener listener;
-
-    private boolean released;
-
-    public ListenerHolder(Player.EventListener listener) {
-      this.listener = listener;
-    }
-
-    /** Prevents any further {@link ListenerInvocation} to be executed on {@link #listener}. */
-    public void release() {
-      released = true;
-    }
-
-    /**
-     * Executes the given {@link ListenerInvocation} on {@link #listener}. Does nothing if {@link
-     * #release} has been called on this instance.
-     */
-    public void invoke(ListenerInvocation listenerInvocation) {
-      if (!released) {
-        listenerInvocation.invokeListener(listener);
-      }
-    }
-
-    @Override
-    public boolean equals(@Nullable Object other) {
-      if (this == other) {
-        return true;
-      }
-      if (other == null || getClass() != other.getClass()) {
-        return false;
-      }
-      return listener.equals(((ListenerHolder) other).listener);
-    }
-
-    @Override
-    public int hashCode() {
-      return listener.hashCode();
-    }
-  }
-
-  /** Parameterized invocation of a {@link Player.EventListener} method. */
-  protected interface ListenerInvocation {
-
-    /** Executes the invocation on the given {@link Player.EventListener}. */
-    void invokeListener(Player.EventListener listener);
-  }
 }

library/core/src/main/java/com/google/android/exoplayer2/ExoPlayerImpl.java

@@ -42,14 +42,13 @@ import com.google.android.exoplayer2.trackselection.TrackSelectorResult;
 import com.google.android.exoplayer2.upstream.BandwidthMeter;
 import com.google.android.exoplayer2.util.Assertions;
 import com.google.android.exoplayer2.util.Clock;
+import com.google.android.exoplayer2.util.ListenerSet;
 import com.google.android.exoplayer2.util.Log;
 import com.google.android.exoplayer2.util.Util;
 import com.google.common.collect.ImmutableList;
-import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeoutException;
 
 /**
@@ -74,9 +73,8 @@ import java.util.concurrent.TimeoutException;
   private final ExoPlayerImplInternal.PlaybackInfoUpdateListener playbackInfoUpdateListener;
   private final ExoPlayerImplInternal internalPlayer;
   private final Handler internalPlayerHandler;
-  private final CopyOnWriteArrayList<ListenerHolder> listeners;
+  private final ListenerSet<Player.EventListener> listeners;
   private final Timeline.Period period;
-  private final ArrayDeque<Runnable> pendingListenerNotifications;
   private final List<MediaSourceHolderSnapshot> mediaSourceHolderSnapshots;
   private final boolean useLazyPreparation;
   private final MediaSourceFactory mediaSourceFactory;
@@ -152,7 +150,7 @@ import java.util.concurrent.TimeoutException;
     this.pauseAtEndOfMediaItems = pauseAtEndOfMediaItems;
     this.applicationLooper = applicationLooper;
     repeatMode = Player.REPEAT_MODE_OFF;
-    listeners = new CopyOnWriteArrayList<>();
+    listeners = new ListenerSet<>();
     mediaSourceHolderSnapshots = new ArrayList<>();
     shuffleOrder = new ShuffleOrder.DefaultShuffleOrder(/* length= */ 0);
     emptyTrackSelectorResult =
@@ -167,7 +165,6 @@ import java.util.concurrent.TimeoutException;
         playbackInfoUpdate ->
             playbackInfoUpdateHandler.post(() -> handlePlaybackInfo(playbackInfoUpdate));
     playbackInfo = PlaybackInfo.createDummy(emptyTrackSelectorResult);
-    pendingListenerNotifications = new ArrayDeque<>();
     if (analyticsCollector != null) {
       analyticsCollector.setPlayer(this);
       addListener(analyticsCollector);
@@ -255,18 +252,12 @@ import java.util.concurrent.TimeoutException;
 
   @Override
   public void addListener(Player.EventListener listener) {
-    Assertions.checkNotNull(listener);
+    listeners.add(listener);
-    listeners.addIfAbsent(new ListenerHolder(listener));
   }
 
   @Override
   public void removeListener(Player.EventListener listener) {
-    for (ListenerHolder listenerHolder : listeners) {
+    listeners.remove(listener);
-      if (listenerHolder.listener.equals(listener)) {
-        listenerHolder.release();
-        listeners.remove(listenerHolder);
-      }
-    }
   }
 
   @Override
@@ -560,7 +551,7 @@ import java.util.concurrent.TimeoutException;
     if (this.repeatMode != repeatMode) {
       this.repeatMode = repeatMode;
       internalPlayer.setRepeatMode(repeatMode);
-      notifyListeners(listener -> listener.onRepeatModeChanged(repeatMode));
+      listeners.sendEvent(listener -> listener.onRepeatModeChanged(repeatMode));
     }
   }
 
@@ -574,7 +565,7 @@ import java.util.concurrent.TimeoutException;
     if (this.shuffleModeEnabled != shuffleModeEnabled) {
       this.shuffleModeEnabled = shuffleModeEnabled;
       internalPlayer.setShuffleModeEnabled(shuffleModeEnabled);
-      notifyListeners(listener -> listener.onShuffleModeEnabledChanged(shuffleModeEnabled));
+      listeners.sendEvent(listener -> listener.onShuffleModeEnabledChanged(shuffleModeEnabled));
     }
   }
 
@@ -723,7 +714,7 @@ import java.util.concurrent.TimeoutException;
         + ExoPlayerLibraryInfo.VERSION_SLASHY + "] [" + Util.DEVICE_DEBUG_INFO + "] ["
         + ExoPlayerLibraryInfo.registeredModules() + "]");
     if (!internalPlayer.release()) {
-      notifyListeners(
+      listeners.sendEvent(
           listener ->
               listener.onPlayerError(
                   ExoPlaybackException.createForTimeout(
@@ -942,6 +933,8 @@ import java.util.concurrent.TimeoutException;
     }
   }
 
+  // Calling deprecated listeners.
+  @SuppressWarnings("deprecation")
   private void updatePlaybackInfo(
       PlaybackInfo playbackInfo,
       boolean positionDiscontinuity,
@@ -949,39 +942,106 @@ import java.util.concurrent.TimeoutException;
       @TimelineChangeReason int timelineChangeReason,
       @PlayWhenReadyChangeReason int playWhenReadyChangeReason,
       boolean seekProcessed) {
-    // Assign playback info immediately such that all getters return the right values.
+    // Assign playback info immediately such that all getters return the right values, but keep
+    // snapshot of previous and new state so that listener invocations are triggered correctly.
     PlaybackInfo previousPlaybackInfo = this.playbackInfo;
+    PlaybackInfo newPlaybackInfo = playbackInfo;
     this.playbackInfo = playbackInfo;
 
     Pair<Boolean, Integer> mediaItemTransitionInfo =
         evaluateMediaItemTransitionReason(
-            playbackInfo,
+            newPlaybackInfo,
             previousPlaybackInfo,
             positionDiscontinuity,
             positionDiscontinuityReason,
-            !previousPlaybackInfo.timeline.equals(playbackInfo.timeline));
+            !previousPlaybackInfo.timeline.equals(newPlaybackInfo.timeline));
     boolean mediaItemTransitioned = mediaItemTransitionInfo.first;
     int mediaItemTransitionReason = mediaItemTransitionInfo.second;
-    @Nullable MediaItem newMediaItem = null;
+    if (!previousPlaybackInfo.timeline.equals(newPlaybackInfo.timeline)) {
-    if (mediaItemTransitioned && !playbackInfo.timeline.isEmpty()) {
+      listeners.queueEvent(
-      int windowIndex =
+          listener -> listener.onTimelineChanged(newPlaybackInfo.timeline, timelineChangeReason));
-          playbackInfo.timeline.getPeriodByUid(playbackInfo.periodId.periodUid, period).windowIndex;
-      newMediaItem = playbackInfo.timeline.getWindow(windowIndex, window).mediaItem;
     }
-    notifyListeners(
+    if (positionDiscontinuity) {
-        new PlaybackInfoUpdate(
+      listeners.queueEvent(
-            playbackInfo,
+          listener -> listener.onPositionDiscontinuity(positionDiscontinuityReason));
-            previousPlaybackInfo,
+    }
-            listeners,
+    if (mediaItemTransitioned) {
-            trackSelector,
+      @Nullable final MediaItem mediaItem;
-            positionDiscontinuity,
+      if (!newPlaybackInfo.timeline.isEmpty()) {
-            positionDiscontinuityReason,
+        int windowIndex =
-            timelineChangeReason,
+            newPlaybackInfo.timeline.getPeriodByUid(newPlaybackInfo.periodId.periodUid, period)
-            mediaItemTransitioned,
+                .windowIndex;
-            mediaItemTransitionReason,
+        mediaItem = newPlaybackInfo.timeline.getWindow(windowIndex, window).mediaItem;
-            newMediaItem,
+      } else {
-            playWhenReadyChangeReason,
+        mediaItem = null;
-            seekProcessed));
+      }
+      listeners.queueEvent(
+          listener -> listener.onMediaItemTransition(mediaItem, mediaItemTransitionReason));
+    }
+    if (previousPlaybackInfo.playbackError != newPlaybackInfo.playbackError
+        && newPlaybackInfo.playbackError != null) {
+      listeners.queueEvent(listener -> listener.onPlayerError(newPlaybackInfo.playbackError));
+    }
+    if (previousPlaybackInfo.trackSelectorResult != newPlaybackInfo.trackSelectorResult) {
+      trackSelector.onSelectionActivated(newPlaybackInfo.trackSelectorResult.info);
+      listeners.queueEvent(
+          listener ->
+              listener.onTracksChanged(
+                  newPlaybackInfo.trackGroups, newPlaybackInfo.trackSelectorResult.selections));
+    }
+    if (!previousPlaybackInfo.staticMetadata.equals(newPlaybackInfo.staticMetadata)) {
+      listeners.queueEvent(
+          listener -> listener.onStaticMetadataChanged(newPlaybackInfo.staticMetadata));
+    }
+    if (previousPlaybackInfo.isLoading != newPlaybackInfo.isLoading) {
+      listeners.queueEvent(listener -> listener.onIsLoadingChanged(newPlaybackInfo.isLoading));
+    }
+    if (previousPlaybackInfo.playbackState != newPlaybackInfo.playbackState
+        || previousPlaybackInfo.playWhenReady != newPlaybackInfo.playWhenReady) {
+      listeners.queueEvent(
+          listener ->
+              listener.onPlayerStateChanged(
+                  newPlaybackInfo.playWhenReady, newPlaybackInfo.playbackState));
+    }
+    if (previousPlaybackInfo.playbackState != newPlaybackInfo.playbackState) {
+      listeners.queueEvent(
+          listener -> listener.onPlaybackStateChanged(newPlaybackInfo.playbackState));
+    }
+    if (previousPlaybackInfo.playWhenReady != newPlaybackInfo.playWhenReady) {
+      listeners.queueEvent(
+          listener ->
+              listener.onPlayWhenReadyChanged(
+                  newPlaybackInfo.playWhenReady, playWhenReadyChangeReason));
+    }
+    if (previousPlaybackInfo.playbackSuppressionReason
+        != newPlaybackInfo.playbackSuppressionReason) {
+      listeners.queueEvent(
+          listener ->
+              listener.onPlaybackSuppressionReasonChanged(
+                  newPlaybackInfo.playbackSuppressionReason));
+    }
+    if (isPlaying(previousPlaybackInfo) != isPlaying(newPlaybackInfo)) {
+      listeners.queueEvent(listener -> listener.onIsPlayingChanged(isPlaying(newPlaybackInfo)));
+    }
+    if (!previousPlaybackInfo.playbackParameters.equals(newPlaybackInfo.playbackParameters)) {
+      listeners.queueEvent(
+          listener -> listener.onPlaybackParametersChanged(newPlaybackInfo.playbackParameters));
+    }
+    if (seekProcessed) {
+      listeners.queueEvent(EventListener::onSeekProcessed);
+    }
+    if (previousPlaybackInfo.offloadSchedulingEnabled != newPlaybackInfo.offloadSchedulingEnabled) {
+      listeners.queueEvent(
+          listener ->
+              listener.onExperimentalOffloadSchedulingEnabledChanged(
+                  newPlaybackInfo.offloadSchedulingEnabled));
+    }
+    if (previousPlaybackInfo.sleepingForOffload != newPlaybackInfo.sleepingForOffload) {
+      listeners.queueEvent(
+          listener ->
+              listener.onExperimentalSleepingForOffloadChanged(newPlaybackInfo.sleepingForOffload));
+    }
+    listeners.flushEvents();
   }
 
   private Pair<Boolean, Integer> evaluateMediaItemTransitionReason(
@@ -1332,23 +1392,6 @@ import java.util.concurrent.TimeoutException;
     return timeline.getPeriodPosition(window, period, windowIndex, C.msToUs(windowPositionMs));
   }
 
-  private void notifyListeners(ListenerInvocation listenerInvocation) {
-    CopyOnWriteArrayList<ListenerHolder> listenerSnapshot = new CopyOnWriteArrayList<>(listeners);
-    notifyListeners(() -> invokeAll(listenerSnapshot, listenerInvocation));
-  }
-
-  private void notifyListeners(Runnable listenerNotificationRunnable) {
-    boolean isRunningRecursiveListenerNotification = !pendingListenerNotifications.isEmpty();
-    pendingListenerNotifications.addLast(listenerNotificationRunnable);
-    if (isRunningRecursiveListenerNotification) {
-      return;
-    }
-    while (!pendingListenerNotifications.isEmpty()) {
-      pendingListenerNotifications.peekFirst().run();
-      pendingListenerNotifications.removeFirst();
-    }
-  }
-
   private long periodPositionUsToWindowPositionMs(MediaPeriodId periodId, long positionUs) {
     long positionMs = C.usToMs(positionUs);
     playbackInfo.timeline.getPeriodByUid(periodId.periodUid, period);
@@ -1356,184 +1399,11 @@ import java.util.concurrent.TimeoutException;
     return positionMs;
   }
 
-  private static final class PlaybackInfoUpdate implements Runnable {
-
-    private final PlaybackInfo playbackInfo;
-    private final CopyOnWriteArrayList<ListenerHolder> listenerSnapshot;
-    private final TrackSelector trackSelector;
-    private final boolean positionDiscontinuity;
-    @DiscontinuityReason private final int positionDiscontinuityReason;
-    @TimelineChangeReason private final int timelineChangeReason;
-    private final boolean mediaItemTransitioned;
-    private final int mediaItemTransitionReason;
-    @Nullable private final MediaItem mediaItem;
-    @PlayWhenReadyChangeReason private final int playWhenReadyChangeReason;
-    private final boolean seekProcessed;
-    private final boolean playbackStateChanged;
-    private final boolean playbackErrorChanged;
-    private final boolean isLoadingChanged;
-    private final boolean timelineChanged;
-    private final boolean trackSelectorResultChanged;
-    private final boolean staticMetadataChanged;
-    private final boolean playWhenReadyChanged;
-    private final boolean playbackSuppressionReasonChanged;
-    private final boolean isPlayingChanged;
-    private final boolean playbackParametersChanged;
-    private final boolean offloadSchedulingEnabledChanged;
-    private final boolean sleepingForOffloadChanged;
-
-    public PlaybackInfoUpdate(
-        PlaybackInfo playbackInfo,
-        PlaybackInfo previousPlaybackInfo,
-        CopyOnWriteArrayList<ListenerHolder> listeners,
-        TrackSelector trackSelector,
-        boolean positionDiscontinuity,
-        @DiscontinuityReason int positionDiscontinuityReason,
-        @TimelineChangeReason int timelineChangeReason,
-        boolean mediaItemTransitioned,
-        @MediaItemTransitionReason int mediaItemTransitionReason,
-        @Nullable MediaItem mediaItem,
-        @PlayWhenReadyChangeReason int playWhenReadyChangeReason,
-        boolean seekProcessed) {
-      this.playbackInfo = playbackInfo;
-      this.listenerSnapshot = new CopyOnWriteArrayList<>(listeners);
-      this.trackSelector = trackSelector;
-      this.positionDiscontinuity = positionDiscontinuity;
-      this.positionDiscontinuityReason = positionDiscontinuityReason;
-      this.timelineChangeReason = timelineChangeReason;
-      this.mediaItemTransitioned = mediaItemTransitioned;
-      this.mediaItemTransitionReason = mediaItemTransitionReason;
-      this.mediaItem = mediaItem;
-      this.playWhenReadyChangeReason = playWhenReadyChangeReason;
-      this.seekProcessed = seekProcessed;
-      playbackStateChanged = previousPlaybackInfo.playbackState != playbackInfo.playbackState;
-      playbackErrorChanged =
-          previousPlaybackInfo.playbackError != playbackInfo.playbackError
-              && playbackInfo.playbackError != null;
-      isLoadingChanged = previousPlaybackInfo.isLoading != playbackInfo.isLoading;
-      timelineChanged = !previousPlaybackInfo.timeline.equals(playbackInfo.timeline);
-      trackSelectorResultChanged =
-          previousPlaybackInfo.trackSelectorResult != playbackInfo.trackSelectorResult;
-      staticMetadataChanged =
-          !previousPlaybackInfo.staticMetadata.equals(playbackInfo.staticMetadata);
-      playWhenReadyChanged = previousPlaybackInfo.playWhenReady != playbackInfo.playWhenReady;
-      playbackSuppressionReasonChanged =
-          previousPlaybackInfo.playbackSuppressionReason != playbackInfo.playbackSuppressionReason;
-      isPlayingChanged = isPlaying(previousPlaybackInfo) != isPlaying(playbackInfo);
-      playbackParametersChanged =
-          !previousPlaybackInfo.playbackParameters.equals(playbackInfo.playbackParameters);
-      offloadSchedulingEnabledChanged =
-          previousPlaybackInfo.offloadSchedulingEnabled != playbackInfo.offloadSchedulingEnabled;
-      sleepingForOffloadChanged =
-          previousPlaybackInfo.sleepingForOffload != playbackInfo.sleepingForOffload;
-    }
-
-    @SuppressWarnings("deprecation")
-    @Override
-    public void run() {
-      if (timelineChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener -> listener.onTimelineChanged(playbackInfo.timeline, timelineChangeReason));
-      }
-      if (positionDiscontinuity) {
-        invokeAll(
-            listenerSnapshot,
-            listener -> listener.onPositionDiscontinuity(positionDiscontinuityReason));
-      }
-      if (mediaItemTransitioned) {
-        invokeAll(
-            listenerSnapshot,
-            listener -> listener.onMediaItemTransition(mediaItem, mediaItemTransitionReason));
-      }
-      if (playbackErrorChanged) {
-        invokeAll(listenerSnapshot, listener -> listener.onPlayerError(playbackInfo.playbackError));
-      }
-      if (trackSelectorResultChanged) {
-        trackSelector.onSelectionActivated(playbackInfo.trackSelectorResult.info);
-        invokeAll(
-            listenerSnapshot,
-            listener ->
-                listener.onTracksChanged(
-                    playbackInfo.trackGroups, playbackInfo.trackSelectorResult.selections));
-      }
-      if (staticMetadataChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener -> listener.onStaticMetadataChanged(playbackInfo.staticMetadata));
-      }
-      if (isLoadingChanged) {
-        invokeAll(
-            listenerSnapshot, listener -> listener.onIsLoadingChanged(playbackInfo.isLoading));
-      }
-      if (playbackStateChanged || playWhenReadyChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener ->
-                listener.onPlayerStateChanged(
-                    playbackInfo.playWhenReady, playbackInfo.playbackState));
-      }
-      if (playbackStateChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener -> listener.onPlaybackStateChanged(playbackInfo.playbackState));
-      }
-      if (playWhenReadyChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener ->
-                listener.onPlayWhenReadyChanged(
-                    playbackInfo.playWhenReady, playWhenReadyChangeReason));
-      }
-      if (playbackSuppressionReasonChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener ->
-                listener.onPlaybackSuppressionReasonChanged(
-                    playbackInfo.playbackSuppressionReason));
-      }
-      if (isPlayingChanged) {
-        invokeAll(
-            listenerSnapshot, listener -> listener.onIsPlayingChanged(isPlaying(playbackInfo)));
-      }
-      if (playbackParametersChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener -> {
-              listener.onPlaybackParametersChanged(playbackInfo.playbackParameters);
-            });
-      }
-      if (seekProcessed) {
-        invokeAll(listenerSnapshot, EventListener::onSeekProcessed);
-      }
-      if (offloadSchedulingEnabledChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener ->
-                listener.onExperimentalOffloadSchedulingEnabledChanged(
-                    playbackInfo.offloadSchedulingEnabled));
-      }
-      if (sleepingForOffloadChanged) {
-        invokeAll(
-            listenerSnapshot,
-            listener ->
-                listener.onExperimentalSleepingForOffloadChanged(playbackInfo.sleepingForOffload));
-      }
-    }
-
   private static boolean isPlaying(PlaybackInfo playbackInfo) {
     return playbackInfo.playbackState == Player.STATE_READY
         && playbackInfo.playWhenReady
         && playbackInfo.playbackSuppressionReason == PLAYBACK_SUPPRESSION_REASON_NONE;
   }
-  }
-
-  private static void invokeAll(
-      CopyOnWriteArrayList<ListenerHolder> listeners, ListenerInvocation listenerInvocation) {
-    for (ListenerHolder listenerHolder : listeners) {
-      listenerHolder.invoke(listenerInvocation);
-    }
-  }
 
   private static final class MediaSourceHolderSnapshot implements MediaSourceInfoHolder {
 

library/core/src/main/java/com/google/android/exoplayer2/util/ListenerSet.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright (C) 2020 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.util;
+
+import androidx.annotation.Nullable;
+import java.util.ArrayDeque;
+import java.util.concurrent.CopyOnWriteArraySet;
+import javax.annotation.Nonnull;
+
+/**
+ * A set of listeners.
+ *
+ * <p>Events are guaranteed to arrive in the order in which they happened even if a new event is
+ * triggered recursively from another listener.
+ *
+ * <p>Events are also guaranteed to be only sent to the listeners registered at the time the event
+ * was enqueued and haven't been removed since.
+ *
+ * @param <T> The listener type.
+ */
+public final class ListenerSet<T> {
+
+  /**
+   * An event sent to a listener.
+   *
+   * @param <T> The listener type.
+   */
+  public interface Event<T> {
+
+    /** Invokes the event notification on the given listener. */
+    void invoke(T listener);
+  }
+
+  private final CopyOnWriteArraySet<ListenerHolder<T>> listeners;
+  private final ArrayDeque<Runnable> flushingEvents;
+  private final ArrayDeque<Runnable> queuedEvents;
+
+  /** Creates the listener set. */
+  public ListenerSet() {
+    listeners = new CopyOnWriteArraySet<>();
+    flushingEvents = new ArrayDeque<>();
+    queuedEvents = new ArrayDeque<>();
+  }
+
+  /**
+   * Adds a listener to the set.
+   *
+   * <p>If a listener is already present, it will not be added again.
+   *
+   * @param listener The listener to be added.
+   */
+  public void add(T listener) {
+    Assertions.checkNotNull(listener);
+    listeners.add(new ListenerHolder<T>(listener));
+  }
+
+  /**
+   * Removes a listener from the set.
+   *
+   * <p>If the listener is not present, nothing happens.
+   *
+   * @param listener The listener to be removed.
+   */
+  public void remove(T listener) {
+    for (ListenerHolder<T> listenerHolder : listeners) {
+      if (listenerHolder.listener.equals(listener)) {
+        listenerHolder.release();
+        listeners.remove(listenerHolder);
+      }
+    }
+  }
+
+  /**
+   * Adds an event that is sent to the listeners when {@link #flushEvents} is called.
+   *
+   * @param event The event.
+   */
+  public void queueEvent(Event<T> event) {
+    CopyOnWriteArraySet<ListenerHolder<T>> listenerSnapshot = new CopyOnWriteArraySet<>(listeners);
+    queuedEvents.add(
+        () -> {
+          for (ListenerHolder<T> holder : listenerSnapshot) {
+            holder.invoke(event);
+          }
+        });
+  }
+
+  /** Notifies listeners of events previously enqueued with {@link #queueEvent(Event)}. */
+  public void flushEvents() {
+    boolean recursiveFlushInProgress = !flushingEvents.isEmpty();
+    flushingEvents.addAll(queuedEvents);
+    queuedEvents.clear();
+    if (recursiveFlushInProgress) {
+      // Recursive call to flush. Let the outer call handle the flush queue.
+      return;
+    }
+    while (!flushingEvents.isEmpty()) {
+      flushingEvents.peekFirst().run();
+      flushingEvents.removeFirst();
+    }
+  }
+
+  /**
+   * {@link #queueEvent(Event) Queues} a single event and immediately {@link #flushEvents() flushes}
+   * the event queue to notify all listeners.
+   *
+   * @param event The event.
+   */
+  public void sendEvent(Event<T> event) {
+    queueEvent(event);
+    flushEvents();
+  }
+
+  private static final class ListenerHolder<T> {
+
+    @Nonnull public final T listener;
+
+    private boolean released;
+
+    public ListenerHolder(@Nonnull T listener) {
+      this.listener = listener;
+    }
+
+    public void release() {
+      released = true;
+    }
+
+    public void invoke(Event<T> event) {
+      if (!released) {
+        event.invoke(listener);
+      }
+    }
+
+    @Override
+    public boolean equals(@Nullable Object other) {
+      if (this == other) {
+        return true;
+      }
+      if (other == null || getClass() != other.getClass()) {
+        return false;
+      }
+      return listener.equals(((ListenerHolder<?>) other).listener);
+    }
+
+    @Override
+    public int hashCode() {
+      return listener.hashCode();
+    }
+  }
+}

library/core/src/test/java/com/google/android/exoplayer2/util/ListenerSetTest.java

@@ -0,0 +1,194 @@
+/*
+ * Copyright (C) 2020 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.util;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import androidx.test.ext.junit.runners.AndroidJUnit4;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+
+/** Unit test for {@link ListenerSet}. */
+@RunWith(AndroidJUnit4.class)
+public class ListenerSetTest {
+
+  @Test
+  public void queueEvent_isNotSentWithoutFlush() {
+    ListenerSet<TestListener> listenerSet = new ListenerSet<>();
+    TestListener listener = mock(TestListener.class);
+    listenerSet.add(listener);
+
+    listenerSet.queueEvent(TestListener::callback1);
+    listenerSet.queueEvent(TestListener::callback2);
+
+    verifyNoMoreInteractions(listener);
+  }
+
+  @Test
+  public void flushEvents_sendsPreviouslyQueuedEventsToAllListeners() {
+    ListenerSet<TestListener> listenerSet = new ListenerSet<>();
+    TestListener listener1 = mock(TestListener.class);
+    TestListener listener2 = mock(TestListener.class);
+    listenerSet.add(listener1);
+    listenerSet.add(listener2);
+
+    listenerSet.queueEvent(TestListener::callback1);
+    listenerSet.queueEvent(TestListener::callback2);
+    listenerSet.queueEvent(TestListener::callback1);
+    listenerSet.flushEvents();
+
+    InOrder inOrder = Mockito.inOrder(listener1, listener2);
+    inOrder.verify(listener1).callback1();
+    inOrder.verify(listener2).callback1();
+    inOrder.verify(listener1).callback2();
+    inOrder.verify(listener2).callback2();
+    inOrder.verify(listener1).callback1();
+    inOrder.verify(listener2).callback1();
+    inOrder.verifyNoMoreInteractions();
+  }
+
+  @Test
+  public void flushEvents_recursive_sendsEventsInCorrectOrder() {
+    ListenerSet<TestListener> listenerSet = new ListenerSet<>();
+    // Listener1 sends callback3 recursively when receiving callback1.
+    TestListener listener1 =
+        spy(
+            new TestListener() {
+              @Override
+              public void callback1() {
+                listenerSet.queueEvent(TestListener::callback3);
+                listenerSet.flushEvents();
+              }
+            });
+    TestListener listener2 = mock(TestListener.class);
+    listenerSet.add(listener1);
+    listenerSet.add(listener2);
+
+    listenerSet.queueEvent(TestListener::callback1);
+    listenerSet.queueEvent(TestListener::callback2);
+    listenerSet.flushEvents();
+
+    InOrder inOrder = Mockito.inOrder(listener1, listener2);
+    inOrder.verify(listener1).callback1();
+    inOrder.verify(listener2).callback1();
+    inOrder.verify(listener1).callback2();
+    inOrder.verify(listener2).callback2();
+    inOrder.verify(listener1).callback3();
+    inOrder.verify(listener2).callback3();
+    inOrder.verifyNoMoreInteractions();
+  }
+
+  @Test
+  public void add_withRecursion_onlyReceivesUpdatesForFutureEvents() {
+    ListenerSet<TestListener> listenerSet = new ListenerSet<>();
+    TestListener listener2 = mock(TestListener.class);
+    // Listener1 adds listener2 recursively.
+    TestListener listener1 =
+        spy(
+            new TestListener() {
+              @Override
+              public void callback1() {
+                listenerSet.add(listener2);
+              }
+            });
+
+    listenerSet.sendEvent(TestListener::callback1);
+    listenerSet.add(listener1);
+    // This should add listener2, but the event should not be received yet as it happened before
+    // listener2 was added.
+    listenerSet.sendEvent(TestListener::callback1);
+    listenerSet.sendEvent(TestListener::callback1);
+
+    verify(listener1, times(2)).callback1();
+    verify(listener2).callback1();
+  }
+
+  @Test
+  public void add_withQueueing_onlyReceivesUpdatesForFutureEvents() {
+    ListenerSet<TestListener> listenerSet = new ListenerSet<>();
+    TestListener listener1 = mock(TestListener.class);
+    TestListener listener2 = mock(TestListener.class);
+
+    // This event is flushed after listener2 was added, but shouldn't be sent to listener2 because
+    // the event itself occurred before the listener was added.
+    listenerSet.add(listener1);
+    listenerSet.queueEvent(TestListener::callback2);
+    listenerSet.add(listener2);
+    listenerSet.queueEvent(TestListener::callback2);
+    listenerSet.flushEvents();
+
+    verify(listener1, times(2)).callback2();
+    verify(listener2).callback2();
+  }
+
+  @Test
+  public void remove_withRecursion_stopsReceivingEventsImmediately() {
+    ListenerSet<TestListener> listenerSet = new ListenerSet<>();
+    TestListener listener2 = mock(TestListener.class);
+    // Listener1 removes listener2 recursively.
+    TestListener listener1 =
+        spy(
+            new TestListener() {
+              @Override
+              public void callback1() {
+                listenerSet.remove(listener2);
+              }
+            });
+    listenerSet.add(listener1);
+    listenerSet.add(listener2);
+
+    // Listener2 shouldn't even get this event as it's removed before the event can be invoked.
+    listenerSet.sendEvent(TestListener::callback1);
+    listenerSet.remove(listener1);
+    listenerSet.sendEvent(TestListener::callback1);
+
+    verify(listener1).callback1();
+    verify(listener2, never()).callback1();
+  }
+
+  @Test
+  public void remove_withQueueing_stopsReceivingEventsImmediately() {
+    ListenerSet<TestListener> listenerSet = new ListenerSet<>();
+    TestListener listener1 = mock(TestListener.class);
+    TestListener listener2 = mock(TestListener.class);
+    listenerSet.add(listener1);
+    listenerSet.add(listener2);
+
+    // Listener1 shouldn't even get this event as it's removed before the event can be invoked.
+    listenerSet.queueEvent(TestListener::callback1);
+    listenerSet.remove(listener1);
+    listenerSet.queueEvent(TestListener::callback1);
+    listenerSet.flushEvents();
+
+    verify(listener1, never()).callback1();
+    verify(listener2, times(2)).callback1();
+  }
+
+  private interface TestListener {
+    default void callback1() {}
+
+    default void callback2() {}
+
+    default void callback3() {}
+  }
+}